import "pkg:/source/api/apiResponse.bs"
import "pkg:/source/api/apiTimeout.bs"
import "pkg:/source/constants/timeouts.bs"
import "pkg:/source/utils/misc.bs"
import "pkg:/source/utils/tasks.bs"
' ── Skipping a request nobody is waiting for ────────────────────────────────
'
' The pool has no way to cancel a request. A caller that stops listening (a Task
' STOPped by replaceTask() while it waits in fetchRes, a promise abandoned at
' teardown, a pipeline run ended early, a caller that timed out) leaves its request
' in the coordinator's FIFO queue, where it is still dispatched and holds a pool slot
' for as long as the server takes (up to timeouts.HTTP_MS). On a slow server that
' queues the requests the user IS waiting for behind ones nobody will read: Search
' starts a new SearchTask per keystroke, and the TV guide a new schedule load per page.
'
' So at DISPATCH the coordinator skips a queued request whose caller is gone, and
' completes its result node with `error: "abandoned"` instead. Only reads are skipped:
' a POST or DELETE is something the user asked for (a favorite, a recording, a
' delete), and running it even though the screen moved on is what they want. A POST
' that only reads opts in with `skippable: true` on its request AA
' (BuildGetLiveTvScheduleRequest). A request already on a pool slot is stopped only when
' it is a long one: see "Stopping a request nobody is waiting for" below.
' True when the pool may skip `req` once its caller is gone: a GET or HEAD (a request
' with no method is a GET, executeHttpRequest's pool default), or a request that
' declares `skippable: true`.
function apiRequestIsSkippable(req as dynamic) as boolean
if not isValid(req) then return false
skippable = req.skippable
if type(skippable) = "Boolean" or type(skippable) = "roBoolean"
if skippable then return true
end if
method = req.method
if not isValid(method) or method = "" then return true
method = UCase(method)
return method = "GET" or method = "HEAD"
end function
' True when nobody will read the answer: the caller marked the request abandoned, or
' the Task thread that was waiting for it is no longer running.
function apiRequestCallerGone(abandoned as boolean, owner as dynamic) as boolean
if abandoned then return true
if not isValid(owner) then return false
return not taskThreadIsLive(owner)
end function
' The Task node whose thread is running this code, or invalid on the render or main
' thread. Only fetchRes and a pipeline submit (both Task-thread only) record it.
function poolCallerTask() as dynamic
if not isValid(m.top) then return invalid
if not m.top.isSubtype("Task") then return invalid
return m.top
end function
' ── Stopping a request nobody is waiting for ────────────────────────────────
'
' Skipping only helps a request still in the queue. One already on a slot runs until the
' server answers or its HTTP limit runs out, and a request that asks for a long limit
' (apiTimeout.isLong, such as a library grid's page) can hold its slot for a minute after the user
' has backed out. So when the coordinator SENDS such a request, it watches the Task waiting
' for it, and if that Task stops before the answer arrives it tells the slot to give up
' (apiSlotQuit), which frees the slot at once.
'
' Only long requests are watched. Watching costs the coordinator crossings on the serial
' path every request takes (reading the waiting Task, then subscribing to it, measured at
' 0.5-1.0 ms each on a 512 MB Stick (2026-09-25)), and a request with the default limit
' frees its slot within timeouts.HTTP_MS anyway. Only reads, for the same reason as skipping:
' a write the user asked for should land.
'
' One subscription per Task, not per request: unobserveField drops every observer of the
' field, so a Task with two long requests on slots would lose its watch on the second the
' moment the first was answered. The coordinator subscribes when the first of a Task's long
' requests is sent, unsubscribes when the last is answered or stopped, and a Task that stops
' stops all of them (apiPoolSlotsWatchedBy).
' True when the coordinator watches `req`'s caller and stops `req` if the caller goes away:
' a read (apiRequestIsSkippable) with a longer-than-default HTTP limit.
function apiRequestIsStoppable(req as dynamic) as boolean
return apiTimeout.isLong(req) and apiRequestIsSkippable(req)
end function
' The slots whose requests `caller` is waiting for, [] when none are: each answer was
' already delivered, or no request of its was watched. More than one when a Task has several
' long requests on slots at once (an apiPipeline run), which share one subscription to it.
'
' @param inFlight - one element per pool slot: invalid = free, else an entry whose
' `watchedCaller` is the Task the coordinator watches for it, if any
' @param caller - a Task node
function apiPoolSlotsWatchedBy(inFlight as object, caller as object) as object
slots = []
for i = 0 to inFlight.count() - 1
entry = inFlight[i]
if isValid(entry) and isValid(entry.watchedCaller)
if entry.watchedCaller.isSameNode(caller) then slots.push(i)
end if
end for
return slots
end function
' Tells a pool slot to give up the request it is waiting on. roku-requests reads
' `m.top.quit` on every pass of its wait and cancels the transfer once it is true.
'
' The field exists only while a stop is pending, so ApiTask.xml must not declare it:
' roku-requests reads it on every pass of every request in the app, and a read of a field
' the node has costs about twice a read of one it lacks (docs/architecture/api.md, "A long
' request already on a slot"). Added here, removed by apiSlotClearQuit before the slot's
' next request.
sub apiSlotQuit(slot as object)
if not slot.hasField("quit") then slot.addField("quit", "boolean", false)
slot.quit = true
end sub
' Ends a stop apiSlotQuit started. Called before the slot is sent its next request, which
' a `quit` still set would stop too. Safe on a slot that was never told to quit.
sub apiSlotClearQuit(slot as object)
if slot.hasField("quit") then slot.removeField("quit")
end sub
' Marks a request as one its caller has stopped waiting for, so the coordinator can
' skip it if it has not been dispatched yet. Safe on invalid and on a finished request.
sub abandonApiRequest(resultNode as dynamic)
if isValid(resultNode) then resultNode.abandoned = true
end sub
' Skips the entry when its caller is gone and it is safe to skip, completing its
' result node so the node never looks pending; returns true if it did.
'
' Called by apiPoolNextDispatch, and only for an entry that waited for a slot: every
' read here crosses to the render thread, and the coordinator is the serial path every
' request passes through. Measured on a Stick 4K (2026-09-22, Home load, 96 requests):
' 7796 us per entry checked, against 1559 us for the slot write beside it.
'
' Reads the request AA the coordinator already holds before touching the node, so a
' request that can never be skipped costs no cross-thread reads at all.
'
' @param entry - { req: request AA, resultNode: ApiResultNode }
function settleIfCallerGone(entry as object) as boolean
if not apiRequestIsSkippable(entry.req) then return false
node = entry.resultNode
if not isValid(node) then return false
owner = node.owner
if not apiRequestCallerGone(node.abandoned, owner) then return false
' Skipped, so this Task reference is known dead: drop it. A DISPATCHED entry keeps
' its owner: that Task is still waiting for the answer, and clearing it would cost a
' crossing on the path every request takes. The node lets go when it is released,
' which the coordinator's 50-child prune bounds. (Folding the clear into the delivery
' write as setFields({result, owner: invalid}) was measured SLOWER than the plain
' write: 27.4 ms against 10.8, see ADR 0040.)
if isValid(owner) then node.owner = invalid
node.result = {
requestId: entry.req.requestId,
ok: false,
statusCode: 0,
json: invalid,
text: "",
error: apiResponse.ERROR_ABANDONED
}
node.isDone = true
return true
end function
' The coordinator's per-slot decision, as one step: take the next queued entry and say
' what to do with it. Extracted from ApiQueueTask.tryDispatch so tests can drive the
' rules: a skip that claimed a slot would leave that slot marked busy with nothing
' running on it, dead for the rest of the session, and the shell is not reachable.
'
' Mutates `queue` and `inFlight` exactly as the coordinator needs: the returned entry is
' off the queue, and a dispatched one is already recorded in its slot.
'
' @param queue - the FIFO of { req, resultNode, waited } entries
' @param inFlight - one element per pool slot: invalid = free
' @return an AA { slot, entry } naming the slot to send it to, the same AA with slot -1
' for an entry completed as abandoned, or invalid when nothing can move now
function apiPoolNextDispatch(queue as object, inFlight as object) as dynamic
if queue.count() = 0 then return invalid
freeSlot = -1
for i = 0 to inFlight.count() - 1
if not isValid(inFlight[i])
freeSlot = i
exit for
end if
end for
if freeSlot = -1
' Every slot is busy, so everything still queued is now WAITING, the only
' population whose caller can go away before it is dispatched. An entry that
' finds a free slot on arrival is sent in the same breath, so checking it costs
' the crossings and can never find anything.
for each queued in queue
queued.waited = true
end for
return invalid
end if
entry = queue[0]
queue.Delete(0)
' Nested rather than `if entry.waited and settleIfCallerGone(entry)`: the whole point
' is that a request which never waited costs NO cross-thread reads, and that must not
' rest on the evaluation order of `and`.
if entry.waited
if settleIfCallerGone(entry) then return { slot: -1, entry: entry }
end if
inFlight[freeSlot] = entry
return { slot: freeSlot, entry: entry }
end function
' Submits a request to the API queue coordinator and blocks the calling Task thread
' until the response arrives (or apiTimeout.waitMs(req) elapses: timeouts.API_WAIT_MS
' unless the request carries its own `timeoutMs`).
'
' Safe to call concurrently from multiple Task threads: each request gets its own
' ApiResultNode, so responses are never mixed up between callers. The coordinator
' (ApiQueueTask) is the only entity that dispatches to pool slots, eliminating the
' TOCTOU race that existed when callers claimed slots directly.
'
' Delivery mechanism: the ApiResultNode is appended as a CHILD of the coordinator
' node. This is immune to SceneGraph port-event coalescing: even if multiple
' concurrent enqueue wakeup events merge into one, the coordinator reads ALL
' unprocessed children when it wakes up, so no request is ever lost.
'
' @param req - request AA from a Build*Request() method; returns invalid if invalid
' @param requestId - unique string identifying this request (echoed in the response AA)
' @return response AA { requestId, ok, statusCode, json, text } or invalid on timeout (apiTimeout.waitMs)
function fetchRes(req as dynamic, requestId as string) as dynamic
if not isValid(req) then return invalid
apiQueue = m.global.apiQueue
if not isValid(apiQueue) then return invalid
' Startup ordering guard: the coordinator sets isReady=true only AFTER every pool
' slot has registered its request observer AND the coordinator has registered
' its own observers. This guarantees the full pipeline (pool to coordinator to
' callers) is operational before any request flows.
' After the first request this check costs a single field-read rendezvous; it is
' effectively free for the rest of the session once the pipeline is up.
if not apiQueue.isReady
readyPort = CreateObject("roMessagePort")
apiQueue.observeField("isReady", readyPort)
' Double-check after registering: isReady may have become true in the narrow window
' between the first check and observeField(), in which case we skip the wait.
' Loop with short waits: a single 5s wait can miss the event if the coordinator
' becomes ready between observeField and the first wait (event consumed by
' the observe itself on some firmware). Re-checking actual field state each
' iteration makes this immune to event delivery quirks.
for _attempt = 0 to 50
if apiQueue.isReady then exit for
wait(200, readyPort)
end for
apiQueue.unobserveField("isReady")
end if
req.requestId = requestId
' One result node per request: carries the request in and the response out.
' The coordinator reads 'request', dispatches to a pool slot, then writes
' 'result' and sets 'isDone' when the slot completes.
resultNode = CreateObject("roSGNode", "ApiResultNode")
resultNode.request = req
' This Task thread is about to block on the answer. If it is STOPped before a slot
' frees up, the coordinator sees that and skips the request (see above).
if apiRequestIsSkippable(req) then resultNode.owner = poolCallerTask()
' Observe BEFORE appending. If the request completes before we reach wait(),
' the 'isDone' event is already in the port queue and wait() returns immediately.
port = CreateObject("roMessagePort")
resultNode.observeField("isDone", port)
' Append the result node as a child of the coordinator. The coordinator reads
' ALL unprocessed children when woken up. This is immune to event coalescing,
' unlike writing data to a shared field where concurrent writes can merge events.
apiQueue.appendChild(resultNode)
' Wake-up signal: the coordinator ignores the value and reads children instead.
apiQueue.enqueue = requestId
msg = wait(apiTimeout.waitMs(req), port)
if type(msg) = "roSGNodeEvent"
resultNode.unobserveField("isDone")
return resultNode.result
end if
' Timeout: clean up the observer to prevent a stale reference to the dead port.
' The result node stays as a child of the coordinator. Do not removeChild(),
' as that shifts child indices and corrupts the coordinator's m.processedIndex.
' When the coordinator eventually delivers the response it will write to the
' abandoned node (no crash), free the slot, and dispatch the next queued
' request normally. The node is pruned at the 50-child threshold.
resultNode.unobserveField("isDone")
' Still queued (every slot busy for the whole wait)? Then nobody will read it.
abandonApiRequest(resultNode)
return invalid
end function
' Convenience wrapper: submits via the queue and returns the parsed JSON body.
' Returns invalid on timeout, HTTP error, or if req is invalid.
'
' @param req - request AA from a Build*Request() method
' @param requestId - unique string identifying this request
' @return res.json or invalid
function fetchJson(req as dynamic, requestId as string) as dynamic
res = fetchRes(req, requestId)
if isValid(res) and res.ok then return res.json
return invalid
end function
' Submits a request to the API queue coordinator and returns immediately without
' blocking. The caller observes 'isDone' on the returned ApiResultNode via a
' render-thread callback, then reads 'result' for the response.
'
' The render thread does not make the HTTP call. It only creates a node,
' appends it to the coordinator, and returns (~microseconds). The coordinator
' (ApiQueueTask, a Task thread) dispatches to an ApiTask pool slot (another
' Task thread) which performs the actual HTTP request. When done, the result
' is written back to the ApiResultNode and the render-thread observeField
' callback fires.
'
' Use for single API calls with trivial callbacks (set a boolean, read one value).
' Do not use when the callback needs data transforms or array processing:
' use an Orchestrator Task with fetchJson() instead.
'
' Task-thread callers that pipeline several non-blocking requests pass their roMessagePort as
' `port`: isDone is then observed BEFORE the request enters the queue, so a response can never
' land ahead of the observer. Render-thread callers omit it (and should prefer fetchAsync).
'
' ORDERING DEPENDENCY: this function does not wait for the pipeline to come up. Where fetchRes
' blocks up to ~10s on apiQueue.isReady, this returns invalid on the spot, which callers
' reasonably read as "no answer" and degrade on. That is only safe when something that does
' wait has already run: LoadLatestRowsTask, for instance, is reached from Home only after the
' libraries fetch (a blocking fetchRes) has returned, so the pool is provably up by then. A new
' caller that can run before any blocking request must not assume the same.
'
' @param req - request AA from a Build*Request() method; returns invalid if invalid
' @param requestId - unique string identifying this request
' @param port - optional roMessagePort to observe isDone on before enqueueing
' @return ApiResultNode to observe, or invalid if req is invalid or the pipeline is not up yet
function submitApiRequest(req as dynamic, requestId as string, port = invalid as dynamic) as dynamic
if not isValid(req) then return invalid
apiQueue = m.global.apiQueue
if not isValid(apiQueue) or not apiQueue.isReady then return invalid
req.requestId = requestId
resultNode = CreateObject("roSGNode", "ApiResultNode")
resultNode.request = req
if isValid(port)
resultNode.observeField("isDone", port)
' A port means a Task-thread pipeline caller (apiPipeline), which can be STOPped
' while this waits in the queue. Render-thread callers abandon explicitly instead.
if apiRequestIsSkippable(req) then resultNode.owner = poolCallerTask()
end if
apiQueue.appendChild(resultNode)
apiQueue.enqueue = requestId
return resultNode
end function
' Waits for a request a Task thread sent earlier with submitApiRequest(req, id, port), and
' returns its response as fetchRes would have. Lets that Task send a request it needs later,
' do a blocking fetchRes meanwhile, then collect it, so the two waits overlap instead of
' adding up. Waits as long as fetchRes unless the caller sets a shorter limit, and abandons the
' request the same way when that runs out.
'
' @param resultNode - what submitApiRequest returned; invalid (not sent) returns invalid
' @param port - the port passed to submitApiRequest
' @param maxWaitMs - the most to wait, for a caller with its own deadline; invalid waits as
' long as fetchRes would (apiTimeout.waitMs)
' @return the response AA, or invalid when it was not sent or did not answer in time
function collectApiRequest(resultNode as dynamic, port as object, maxWaitMs = invalid as dynamic) as dynamic
if not isValid(resultNode) then return invalid
if not resultNode.isDone
waitMs = apiTimeout.waitMs(resultNode.request)
if isValid(maxWaitMs) and maxWaitMs < waitMs then waitMs = maxWaitMs
' wait(0) blocks forever, so a deadline already spent still gets the shortest real wait.
if waitMs < 1 then waitMs = 1
msg = wait(waitMs, port)
if type(msg) <> "roSGNodeEvent"
resultNode.unobserveField("isDone")
abandonApiRequest(resultNode)
return invalid
end if
end if
resultNode.unobserveField("isDone")
return resultNode.result
end function
' Releases a request a Task thread sent with submitApiRequest(req, id, port) and no longer
' needs, instead of collecting it: stops listening and abandons it.
'
' @param resultNode - what submitApiRequest returned; invalid is a no-op
sub dropApiRequest(resultNode as dynamic)
if not isValid(resultNode) then return
resultNode.unobserveField("isDone")
abandonApiRequest(resultNode)
end sub
' Submits a fire-and-forget request to the global SideEffectTask.
' No response is observed. Calls are serialized, so do not use for cancellable UI requests.
'
' Uses the children-as-vehicle pattern (like submitApiRequest / fetchRes): a FRESH
' ApiResultNode per request means back-to-back submits can never clobber one another.
' The previous single shared 'request' field could: two rapid writes coalesced and a
' POST was silently dropped (issue #744). The 'enqueue' write is only a wake signal; the
' task reads the appended children, not the value.
'
' @param req - request AA: { method, url, headers?, body?, timeoutMs? }
sub SubmitSideEffect(req as dynamic)
if not isValid(req) then return
sideEffectTask = m.global.sideEffectTask
if not isValid(sideEffectTask) then return
' Reuse ApiResultNode purely as a request-carrier vehicle: only its 'request' field is
' read on this path. 'result'/'isDone' stay untouched, since side effects are
' fire-and-forget and deliver no response. (Tier 1 uses all three; see ApiResultNode.xml.)
resultNode = CreateObject("roSGNode", "ApiResultNode")
resultNode.request = req
sideEffectTask.appendChild(resultNode)
' Wake-up only: value ignored (task reads children). alwaysNotify guarantees it fires.
sideEffectTask.enqueue = "submit"
end sub