From 2a065e55e3a081006499c8c02e4e22bb028f5292 Mon Sep 17 00:00:00 2001 From: Richard Higgins Date: Sat, 3 Oct 2026 17:57:05 -1000 Subject: [PATCH] Own request-local native inference cancellation Co-authored-by: Codex (GPT-6) --- .github/workflows/decision-trajectories.yml | 8 ++- README.md | 8 ++- src/bitworld/native_http.nim | 42 +++++++++--- tests/support/artifact_http_probe.nim | 3 +- tests/support/native_http_probe.nim | 7 +- tests/support/native_https_probe.nim | 3 +- .../support/native_request_control_probe.nim | 58 +++++++++++++++++ tests/test_native_request_control.py | 64 +++++++++++++++++++ 8 files changed, 174 insertions(+), 19 deletions(-) create mode 100644 tests/support/native_request_control_probe.nim create mode 100644 tests/test_native_request_control.py diff --git a/.github/workflows/decision-trajectories.yml b/.github/workflows/decision-trajectories.yml index fe0e2ef..a1a3e5e 100644 --- a/.github/workflows/decision-trajectories.yml +++ b/.github/workflows/decision-trajectories.yml @@ -3,9 +3,9 @@ name: Private decision trajectories on: push: branches: [master] - paths: [src/bitworld/native_websocket.nim, tests/support/native_websocket_probe.nim, tests/test_native_websocket.py, src/bitworld/decision_trajectory.nim, src/bitworld/runtime.nim, src/bitworld/native_http.nim, src/bitworld/native_stop.nim, src/bitworld/artifact_runtime.nim, tests/test_artifact_http.py, tests/support/artifact_http_probe.nim, tests/test_native_https.py, tests/support/native_https_probe.nim, tests/test_decision_trajectory.nim, tests/test_native_http.py, tests/support/native_http_probe.nim, .github/workflows/decision-trajectories.yml] + paths: [src/bitworld/native_websocket.nim, tests/support/native_websocket_probe.nim, tests/test_native_websocket.py, src/bitworld/decision_trajectory.nim, src/bitworld/runtime.nim, src/bitworld/native_http.nim, src/bitworld/native_stop.nim, src/bitworld/artifact_runtime.nim, tests/test_artifact_http.py, tests/support/artifact_http_probe.nim, tests/test_native_https.py, tests/support/native_https_probe.nim, tests/test_decision_trajectory.nim, tests/test_native_http.py, tests/support/native_http_probe.nim, tests/support/native_request_control_probe.nim, tests/test_native_request_control.py, .github/workflows/decision-trajectories.yml] pull_request: - paths: [src/bitworld/native_websocket.nim, tests/support/native_websocket_probe.nim, tests/test_native_websocket.py, src/bitworld/decision_trajectory.nim, src/bitworld/runtime.nim, src/bitworld/native_http.nim, src/bitworld/native_stop.nim, src/bitworld/artifact_runtime.nim, tests/test_artifact_http.py, tests/support/artifact_http_probe.nim, tests/test_native_https.py, tests/support/native_https_probe.nim, tests/test_decision_trajectory.nim, tests/test_native_http.py, tests/support/native_http_probe.nim, .github/workflows/decision-trajectories.yml] + paths: [src/bitworld/native_websocket.nim, tests/support/native_websocket_probe.nim, tests/test_native_websocket.py, src/bitworld/decision_trajectory.nim, src/bitworld/runtime.nim, src/bitworld/native_http.nim, src/bitworld/native_stop.nim, src/bitworld/artifact_runtime.nim, tests/test_artifact_http.py, tests/support/artifact_http_probe.nim, tests/test_native_https.py, tests/support/native_https_probe.nim, tests/test_decision_trajectory.nim, tests/test_native_http.py, tests/support/native_http_probe.nim, tests/support/native_request_control_probe.nim, tests/test_native_request_control.py, .github/workflows/decision-trajectories.yml] jobs: test: @@ -37,6 +37,10 @@ jobs: run: | nim c -d:release --threads:on --mm:orc --path:src -o:/tmp/native-http-probe tests/support/native_http_probe.nim python3 tests/test_native_http.py /tmp/native-http-probe + - name: Verify request-local cancellation and later phase admission + run: | + nim c -d:release --threads:on --mm:orc --path:src -o:/tmp/native-request-control-probe tests/support/native_request_control_probe.nim + python3 tests/test_native_request_control.py /tmp/native-request-control-probe - name: Verify bounded private artifact finalization after inference interruption run: | nim c -d:release --threads:on --mm:orc --path:src -o:/tmp/artifact-http-probe tests/support/artifact_http_probe.nim diff --git a/README.md b/README.md index 29bacfd..601dc8c 100644 --- a/README.md +++ b/README.md @@ -205,9 +205,13 @@ coworld certify among_them/coworld_manifest.json ## Native Coworld HTTP lifecycle -`bitworld/native_http.performNativePost(url, headers, body, deadline)` owns one +`bitworld/native_http.performNativePost(url, headers, body, deadline, control)` owns one libcurl handle until cleanup. Pass the same absolute `MonoTime` deadline across -attempts. It returns exact received header/body bytes, observed status, actual +attempts. Each request has an owned `NativeRequestControl`; retain it until every +worker using it joins. `cancelNativeRequest(control)` stops only that request and +returns `nhCanceled`. A later decision uses a fresh control without resetting +the original decision deadline. Global stop remains irreversible. +It returns exact received header/body bytes, observed status, actual transfer completeness, and a typed completion/deadline/interruption/failure kind. A complete transfer arriving after the deadline remains ineligible for selection. diff --git a/src/bitworld/native_http.nim b/src/bitworld/native_http.nim index 906dc42..3e4e782 100644 --- a/src/bitworld/native_http.nim +++ b/src/bitworld/native_http.nim @@ -2,7 +2,7 @@ ## No provider parsing or game acceptance occurs here. All received bytes survive ## timeout/interruption; the handle is cleaned before the result can be sealed. -import std/[monotimes, options, os, posix, times] +import std/[atomics, monotimes, options, os, posix, times] import libcurl except Option import webby/httpheaders import native_stop @@ -15,7 +15,9 @@ type RequestPurpose = enum rpInference, rpArtifact NativeHttpKind* = enum - nhComplete, nhDeadline, nhInterrupted, nhTransportFailure + nhComplete, nhDeadline, nhInterrupted, nhCanceled, nhTransportFailure + NativeRequestControl* = object + canceled: Atomic[bool] NativeHttpResponse* = object kind*: NativeHttpKind httpStatus*: Option[int] @@ -27,6 +29,7 @@ type Transfer = object deadline: MonoTime purpose: RequestPurpose + control: ptr NativeRequestControl headerBytes, bodyBytes: string # The pinned Nim binding omits these existing libcurl options. @@ -36,6 +39,13 @@ const OptProtocols = cast[libcurl.Option](181) OptXferInfoFunction = cast[libcurl.Option](20219) +proc cancelNativeRequest*(control: var NativeRequestControl) {.inline, gcsafe, raises: [].} = + ## Irreversible for this request only; the owner retains it until its worker joins. + control.canceled.store(true, moRelaxed) + +proc nativeRequestCanceled*(control: var NativeRequestControl): bool {.inline, gcsafe, raises: [].} = + control.canceled.load(moRelaxed) + proc requireCurl(code: Code) = if code != E_OK: raise newException(Defect, $easy_strerror(code)) @@ -62,15 +72,19 @@ proc receiveBody(buffer: cstring, size, count: int, context: pointer): int {.cde proc checkTransfer(context: pointer, downloadTotal, downloaded, uploadTotal, uploaded: int64): cint {.cdecl.} = let transfer = cast[ptr Transfer](context) - if (transfer.purpose == rpInference and interruptionRequested()) or + if (transfer.purpose == rpInference and + (interruptionRequested() or transfer.control[].nativeRequestCanceled())) or getMonoTime() >= transfer.deadline: 1 else: 0 proc performOwnedRequest(url: string, httpMethod: ArtifactHttpMethod, headers: HttpHeaders, body: string, deadline: MonoTime, - purpose: RequestPurpose): NativeHttpResponse = + purpose: RequestPurpose, control: var NativeRequestControl): NativeHttpResponse = if purpose == rpInference and interruptionRequested(): result.kind = nhInterrupted return + if purpose == rpInference and control.nativeRequestCanceled(): + result.kind = nhCanceled + return let remaining = (deadline - getMonoTime()).inNanoseconds if remaining <= 0: result.kind = nhDeadline @@ -79,7 +93,7 @@ proc performOwnedRequest(url: string, httpMethod: ArtifactHttpMethod, let handle = easy_init() doAssert handle != nil, "Cannot allocate native HTTP handle" var headerList: Pslist - var transfer = Transfer(deadline: deadline, purpose: purpose) + var transfer = Transfer(deadline: deadline, purpose: purpose, control: control.addr) var oldMask, pipeMask, previousPending: Sigset doAssert sigemptyset(pipeMask) == 0 doAssert sigaddset(pipeMask, SIGPIPE) == 0 @@ -111,8 +125,11 @@ proc performOwnedRequest(url: string, httpMethod: ArtifactHttpMethod, requireCurl(handle.easy_setopt(OptXferInfoFunction, checkTransfer)) let started = getMonoTime() let finalRemaining = (deadline - started).inNanoseconds - if finalRemaining <= 0 or (purpose == rpInference and interruptionRequested()): - result.kind = if purpose == rpInference and interruptionRequested(): nhInterrupted else: nhDeadline + if finalRemaining <= 0 or (purpose == rpInference and + (interruptionRequested() or control.nativeRequestCanceled())): + result.kind = if purpose == rpInference and interruptionRequested(): nhInterrupted + elif purpose == rpInference and control.nativeRequestCanceled(): nhCanceled + else: nhDeadline return let milliseconds = clong((finalRemaining + 999_999) div 1_000_000) requireCurl(handle.easy_setopt(OptTimeoutMs, milliseconds)) @@ -124,6 +141,7 @@ proc performOwnedRequest(url: string, httpMethod: ArtifactHttpMethod, requireCurl(handle.easy_getinfo(INFO_RESPONSE_CODE, status.addr)) if status != 0: result.httpStatus = some(int(status)) if purpose == rpInference and interruptionRequested(): result.kind = nhInterrupted + elif purpose == rpInference and control.nativeRequestCanceled(): result.kind = nhCanceled elif code == E_OPERATION_TIMEOUTED or getMonoTime() >= deadline: result.kind = nhDeadline elif code == E_OK: result.kind = nhComplete else: result.kind = nhTransportFailure @@ -139,17 +157,21 @@ proc performOwnedRequest(url: string, httpMethod: ArtifactHttpMethod, doAssert sigwait(pipeMask, received) == 0 var discardedMask: Sigset doAssert pthread_sigmask(SIG_SETMASK, oldMask, discardedMask) == 0 + if purpose == rpInference and interruptionRequested(): result.kind = nhInterrupted + elif purpose == rpInference and control.nativeRequestCanceled(): result.kind = nhCanceled + elif getMonoTime() >= deadline: result.kind = nhDeadline result.headerBytes = move transfer.headerBytes result.bodyBytes = move transfer.bodyBytes result.responseReaderJoined = some(true) proc performNativePost*(url: string, headers: HttpHeaders, body: string, - deadline: MonoTime): NativeHttpResponse = + deadline: MonoTime, control: var NativeRequestControl): NativeHttpResponse = ## A caller shares one deadline across retries. Never reset it per attempt. - performOwnedRequest(url, ahPost, headers, body, deadline, rpInference) + performOwnedRequest(url, ahPost, headers, body, deadline, rpInference, control) proc performArtifactUpload*(url: string, httpMethod: ArtifactHttpMethod, headers: HttpHeaders, body: string, cleanupDeadline: MonoTime): NativeHttpResponse = ## Checkpoint finalization has its own finite cleanup lifetime after inference stops. ## No caller can disable interruption in the inference API. - performOwnedRequest(url, httpMethod, headers, body, cleanupDeadline, rpArtifact) + var control: NativeRequestControl + performOwnedRequest(url, httpMethod, headers, body, cleanupDeadline, rpArtifact, control) diff --git a/tests/support/artifact_http_probe.nim b/tests/support/artifact_http_probe.nim index 2d15551..872f810 100644 --- a/tests/support/artifact_http_probe.nim +++ b/tests/support/artifact_http_probe.nim @@ -1,11 +1,12 @@ import std/[base64, json, monotimes, options, os, strutils, times] import bitworld/[artifact_runtime, decision_trajectory, native_http, native_stop, runtime] +var control: NativeRequestControl let args = commandLineParams() installNativeStopHandlers() let deadline = getMonoTime() + initDuration(milliseconds = args[2].parseBiggestInt()) if args[1] == "stop": - let inference = performNativePost(args[0] & "/inference", @[], "started", deadline) + let inference = performNativePost(args[0] & "/inference", @[], "started", deadline, control) doAssert inference.kind == nhInterrupted echo $(%*{"inference": $inference.kind}) flushFile(stdout) diff --git a/tests/support/native_http_probe.nim b/tests/support/native_http_probe.nim index d1521d5..108df57 100644 --- a/tests/support/native_http_probe.nim +++ b/tests/support/native_http_probe.nim @@ -1,11 +1,12 @@ import std/[base64, json, monotimes, options, os, strutils, times] import bitworld/[native_http, native_stop] +var control: NativeRequestControl let args = commandLineParams() installNativeStopHandlers() let deadline = getMonoTime() + initDuration(milliseconds = args[1].parseBiggestInt()) let response = performNativePost(args[0], @[("content-type", "application/json")], - "{\"fixture\":true}", deadline) + "{\"fixture\":true}", deadline, control) doAssert response.responseReaderJoined == some(true) echo $(%*{"kind": $response.kind, "status": (if response.httpStatus.isSome: %response.httpStatus.get() else: newJNull()), "headers_b64": encode(response.headerBytes), "body_b64": encode(response.bodyBytes), @@ -13,7 +14,7 @@ echo $(%*{"kind": $response.kind, "status": (if response.httpStatus.isSome: %res "latency_ms": (if response.latencyMs.isSome: %response.latencyMs.get() else: newJNull())}) if args.len == 3: let repeated = performNativePost(args[0], @[("content-type", "application/json")], - "{\"fixture\":true}", deadline) + "{\"fixture\":true}", deadline, control) doAssert repeated.kind == response.kind doAssert repeated.latencyMs.isNone and repeated.httpStatus.isNone doAssert repeated.responseReaderJoined.isNone @@ -21,7 +22,7 @@ if args.len == 3: requestNativeStop() doAssert interruptionRequested() let stopped = performNativePost(args[0], @[("content-type", "application/json")], - "{\"fixture\":true}", getMonoTime() + initDuration(seconds = 5)) + "{\"fixture\":true}", getMonoTime() + initDuration(seconds = 5), control) doAssert stopped.kind == nhInterrupted doAssert stopped.latencyMs.isNone and stopped.httpStatus.isNone doAssert stopped.bodyBytes.len == 0 and stopped.headerBytes.len == 0 diff --git a/tests/support/native_https_probe.nim b/tests/support/native_https_probe.nim index f8895aa..62da154 100644 --- a/tests/support/native_https_probe.nim +++ b/tests/support/native_https_probe.nim @@ -1,10 +1,11 @@ import std/[json, monotimes, options, os, times] import bitworld/native_http +var control: NativeRequestControl let args = commandLineParams() let deadline = getMonoTime() + initDuration(seconds = 3) let response = if args[1] == "inference": - performNativePost(args[0], @[], "private fixture", deadline) + performNativePost(args[0], @[], "private fixture", deadline, control) else: performArtifactUpload(args[0], ahPut, @[], "private fixture", deadline) echo $(%*{"kind": $response.kind, "complete": response.transferComplete, diff --git a/tests/support/native_request_control_probe.nim b/tests/support/native_request_control_probe.nim new file mode 100644 index 0000000..5d0f3c1 --- /dev/null +++ b/tests/support/native_request_control_probe.nim @@ -0,0 +1,58 @@ +## Parent admits cancellation only after the real HTTP fixture receives a request. +import std/[base64, json, monotimes, options, os, times] +import bitworld/[native_http, native_stop] + +type OwnedRequest = object + control: NativeRequestControl + deadline: MonoTime + url: cstring + responseBytes: pointer + responseLen: int + +proc request(job: ptr OwnedRequest) {.thread.} = + let response = performNativePost($job.url, @[], "first", job.deadline, job.control) + let resultBytes = $(%*{"kind": $response.kind, + "status": response.httpStatus, "headers_b64": encode(response.headerBytes), + "body_b64": encode(response.bodyBytes), "complete": response.transferComplete, + "reader_joined": response.responseReaderJoined}) + job.responseLen = resultBytes.len + job.responseBytes = allocShared(resultBytes.len) + doAssert job.responseBytes != nil + copyMem(job.responseBytes, resultBytes[0].unsafeAddr, resultBytes.len) + +let url = commandLineParams()[0] +installNativeStopHandlers() +let deadline = getMonoTime() + initDuration(seconds = 5) +var job = OwnedRequest(deadline: deadline, url: url.cstring) +var worker: Thread[ptr OwnedRequest] +var joined = false +createThread(worker, request, job.addr) +try: + doAssert stdin.readLine() == "cancel" + cancelNativeRequest(job.control) + joinThread(worker) + joined = true + var bytes = newString(job.responseLen) + copyMem(bytes[0].addr, job.responseBytes, bytes.len) + let canceled = parseJson(bytes) + doAssert canceled["kind"].getStr() == "nhCanceled" + doAssert canceled["reader_joined"].getBool() + doAssert not interruptionRequested() + let repeated = performNativePost(url, @[], "not admitted", deadline, job.control) + doAssert repeated.kind == nhCanceled and repeated.httpStatus.isNone + doAssert repeated.responseReaderJoined.isNone + var nextControl: NativeRequestControl + let next = performNativePost(url, @[], "second", deadline, nextControl) + doAssert next.kind == nhComplete and next.responseReaderJoined == some(true) + doAssert next.bodyBytes == "\x00\xffok" + requestNativeStop() + var stoppedControl: NativeRequestControl + let stopped = performNativePost(url, @[], "not admitted", deadline, stoppedControl) + doAssert stopped.kind == nhInterrupted and stopped.httpStatus.isNone + echo $(%*{"canceled": canceled, "later_call_complete": true, + "original_deadline_retained": true, "global_stop_blocks_fresh_control": true}) +finally: + if not joined: + cancelNativeRequest(job.control) + joinThread(worker) + if job.responseBytes != nil: deallocShared(job.responseBytes) diff --git a/tests/test_native_request_control.py b/tests/test_native_request_control.py new file mode 100644 index 0000000..4865bc2 --- /dev/null +++ b/tests/test_native_request_control.py @@ -0,0 +1,64 @@ +"""Real request admission, partial invalid bytes, request-local cancel, and later calls.""" +import base64 +import http.server +import json +import subprocess +import sys +import threading +import time +from pathlib import Path + +probe = Path(sys.argv[1]).resolve() +for trial in range(8): + entered = threading.Event() + release = threading.Event() + requests = [] + + class Fixture(http.server.BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])) + requests.append(body) + assert body in {b"first", b"second"} + self.send_response(200) + self.send_header("X-Fixture", "first") + self.send_header("X-Fixture", "second") + self.send_header("Content-Length", "4") + self.end_headers() + self.wfile.write(b"\xe2\x82" if body == b"first" else b"\x00\xffok") + self.wfile.flush() + if body == b"first": + entered.set() + release.wait(5) + + def log_message(self, *_args): + pass + + with http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fixture) as server: + owner = threading.Thread(target=server.serve_forever) + owner.start() + process = subprocess.Popen([str(probe), f"http://127.0.0.1:{server.server_port}/v1/messages"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + try: + assert entered.wait(3), "the real owned request never entered" + time.sleep(0.2) + started = time.monotonic() + stdout, stderr = process.communicate("cancel\n", timeout=3) + assert process.returncode == 0, stderr + result = json.loads(stdout) + assert result["canceled"]["status"] == 200 + assert not result["canceled"]["complete"] + assert base64.b64decode(result["canceled"]["body_b64"], validate=True) == b"\xe2\x82" + assert b"X-Fixture: first\r\nX-Fixture: second\r\n" in base64.b64decode(result["canceled"]["headers_b64"], validate=True) + assert result["later_call_complete"] and result["original_deadline_retained"] + assert result["global_stop_blocks_fresh_control"] + assert requests == [b"first", b"second"] + print(trial, "phase cancel joined; later call completes; global stop seals", time.monotonic() - started, flush=True) + finally: + if process.poll() is None: + process.terminate() + process.wait(timeout=3) + release.set() + server.shutdown() + owner.join(timeout=3) + assert not owner.is_alive()