From b6ee1c8cd58df009cf98c1b632602a28cb4f3bb7 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Wed, 16 Sep 2026 13:58:00 -0400 Subject: [PATCH 1/5] proxy server add synchronous /cuopt/cuopt endpoint Signed-off-by: Trevor McKay --- .../cuopt_server/proxy_webserver.py | 262 +++++++++++++++--- .../tests/test_grpc_http_proxy.py | 131 ++++++++- 2 files changed, 352 insertions(+), 41 deletions(-) diff --git a/python/cuopt_server/cuopt_server/proxy_webserver.py b/python/cuopt_server/cuopt_server/proxy_webserver.py index fd0ab33a71..dcbca5f74e 100644 --- a/python/cuopt_server/cuopt_server/proxy_webserver.py +++ b/python/cuopt_server/cuopt_server/proxy_webserver.py @@ -36,14 +36,19 @@ IncumbentSolutionResponse, LogResponse, LogResponseModel, + ManagedRequestResponse, RequestResponse, RequestStatusModel, SolutionResponse, ValidationErrorResponse, + cuoptDataInternal, + cuoptdataschema, lp_example_data, lp_msgpack_example_data, lp_zlib_example_data, lpschema, + managed_lp_example_data, + managed_vrp_example_data, ) from cuopt_server.utils.exceptions import ( exception_handler, @@ -87,6 +92,7 @@ validate_file_path, write_result_file, ) +from cuopt_server.utils.logutil import set_ncaid, set_requestid from cuopt_server.utils.routing.conversion import ( create_data_model as create_routing_data_model, create_solver as create_routing_solver, @@ -553,6 +559,29 @@ def _deserialize_convert_submit( data = load_optimization_file(file_path, warnings) else: data = deserialize(ctype, buf) + return _convert_and_submit( + data, + warnings, + validation_only, + incumbent_solutions, + solver_logs, + accept, + result_file, + initial_ids, + ) + + +def _convert_and_submit( + data, + warnings, + validation_only, + incumbent_solutions, + solver_logs, + accept, + result_file, + initial_ids, +): + """Convert a decoded problem body and submit it over gRPC.""" if _looks_like_routing(data): initials = _collect_vrp_initials(initial_ids) data_model, solver_settings, vehicle_ids, task_ids = _prepare_vrp( @@ -904,6 +933,44 @@ def deleterequest( return encode(exception_handler(e), accept) +def _result_envelope(job_id, meta, kind, req_id=""): + """Build the legacy solution envelope for a finished job. + + Returns ``(None, [], [])`` when the result is not available yet. + """ + result_kind, sol = _result_for_job(job_id, meta, kind) + if sol is None: + return None, [], [] + notes = [] + warnings = [] if meta is None else list(meta.get("warnings") or []) + solve_time = 0 + if result_kind == "vrp": + inner = routing_solution_to_http( + sol, + vehicle_ids=None if meta is None else meta.get("vehicle_ids"), + task_ids=None if meta is None else meta.get("task_ids"), + ) + if inner.get("status") == 1: + notes.append(sol.get("status_message") or "") + notes = [n for n in notes if n] + else: + inner = solution_to_http(sol, include_warmstart=False) + try: + notes.append(sol.get_termination_reason()) + except Exception: + pass + if inner.get("solution"): + solve_time = inner["solution"].get("solver_time") or 0 + envelope = make_response( + {"solver_response": inner}, + warnings=warnings, + notes=notes, + reqId=req_id, + total_solve_time=solve_time, + ) + return envelope, warnings, notes + + @app.get( "/cuopt/solution/{id}/warmstart", include_in_schema=False, @@ -955,44 +1022,9 @@ def getsolution( status_code=409, detail=f"job {id} {_status_name(status).lower()}", ) - result_kind, sol = _result_for_job(id, meta, kind) - if sol is None: + envelope, warnings, notes = _result_envelope(id, meta, kind, req_id=id) + if envelope is None: return encode({"reqId": id}, accept) - notes = [] - warnings = [] if meta is None else list(meta.get("warnings") or []) - if result_kind == "vrp": - inner = routing_solution_to_http( - sol, - vehicle_ids=None if meta is None else meta.get("vehicle_ids"), - task_ids=None if meta is None else meta.get("task_ids"), - ) - if inner.get("status") == 1: - notes.append(sol.get("status_message") or "") - solve_time = 0 - envelope = make_response( - {"solver_response": inner}, - warnings=warnings, - notes=[n for n in notes if n], - reqId=id, - total_solve_time=solve_time, - ) - else: - _store_warmstart(id, _warmstart_dict_from_sol(sol)) - inner = solution_to_http(sol, include_warmstart=False) - try: - notes.append(sol.get_termination_reason()) - except Exception: - pass - solve_time = 0 - if inner.get("solution"): - solve_time = inner["solution"].get("solver_time") or 0 - envelope = make_response( - {"solver_response": inner}, - warnings=warnings, - notes=notes, - reqId=id, - total_solve_time=solve_time, - ) resultdir, maxresult, mode = settings.get_result_dir() result_file = "" if meta is None else meta.get("result_file") or "" if result_file and resultdir: @@ -1039,11 +1071,163 @@ def getrequest( return encode(exception_handler(e), accept) +def _submit_wait_solution(ctype, buf, accept): + """Submit, wait, and return a solution for POST /cuopt/cuopt. + + The managed endpoint is stateless: the gRPC job and the proxy metadata + are released before returning, so there is nothing left to poll or + delete afterwards. + """ + body = deserialize(ctype, buf) + try: + wrapper = cuoptDataInternal.parse_obj(body) + except (RequestValidationError, ValidationError): + raise + except Exception as e: + raise HTTPException( + status_code=422, + detail="unable to validate optimization data stream, %s" % str(e), + ) + + if wrapper.data is None: + # NVCF asset files are the only way legacy allowed a null body here + raise HTTPException( + status_code=422, + detail="data is required, NVCF assets are not supported", + ) + + warnings = check_client_version(wrapper.client_version or "") + validation_only = (wrapper.action or "").endswith("Validator") + + job_id = _convert_and_submit( + wrapper.data, + warnings, + validation_only, + False, + False, + accept, + "", + None, + ) + meta = _get_job(job_id) + if meta is not None and meta.get("validation_only"): + envelope = dict(meta["validation_result"]) + envelope.pop("reqId", None) + _pop_job(job_id) + return envelope + + kind = None if meta is None else meta.get("kind") + try: + get_grpc_client().wait(job_id) + status = get_grpc_client().status(job_id) + if not _is_status(status, "COMPLETED"): + raise HTTPException( + status_code=409, + detail=f"job {_status_name(status).lower()}", + ) + envelope, _warnings, _notes = _result_envelope(job_id, meta, kind) + if envelope is None: + raise HTTPException( + status_code=500, detail="solver returned no solution" + ) + return envelope + finally: + try: + get_grpc_client().delete(job_id) + except Exception: + logging.warning(f"could not delete job {job_id}", exc_info=True) + _pop_job(job_id) + + @app.post( "/cuopt/cuopt", + description=( + "Note: This is for the managed service, and users will never call " + "this API directly. Takes all the data and options at once, solves " + "any type of cuOpt problem and returns the result. If you are " + "self-hosting cuOpt, use /cuopt/request instead." + ), + include_in_schema=False, + summary="Managed Service Endpoint", + responses=ManagedRequestResponse, + openapi_extra={ + "requestBody": { + "content": { + "application/json": { + "schema": cuoptdataschema, + "examples": { + "VRP request": {"value": managed_vrp_example_data}, + "LP request": {"value": managed_lp_example_data}, + }, + }, + }, + "required": True, + } + }, ) -async def post_cuopt_sync(): - _not_implemented("POST /cuopt/cuopt") +async def cuopt( + request: Request, + accept: str = Header(default="application/json"), + content_type: str = Header(default="application/json"), + content_length: int = Header(default=0), + nvcf_ncaid: str = Header(default=""), + nvcf_reqid: str = Header(default=""), +): + ctype = content_type + if accept in mime_wild: + accept = mime_json + + try: + await asyncio.to_thread(_require_grpc_healthy) + set_ncaid(nvcf_ncaid) + set_requestid(nvcf_reqid) + + # msgpack is allowed for local testing; NVCF itself only uses json + if ctype not in [mime_json, mime_msgpack]: + raise HTTPException( + status_code=415, + detail=f"Unsupported Content-Type value {ctype}, " + f"supported values are {[mime_json, mime_msgpack]}", + ) + if accept not in [mime_json, mime_msgpack]: + raise HTTPException( + status_code=415, + detail=f"Unsupported Accept value {accept}, " + f"supported values are {[mime_json, mime_msgpack]}", + ) + + sz = int(content_length) + if sz < 0: + raise HTTPException( + status_code=422, detail="Content-Length must be non-negative" + ) + if sz > _max_request_size: + raise HTTPException( + status_code=413, + detail=( + f"Content-Length exceeds maximum of " + f"{_max_request_size} bytes" + ), + ) + if sz == 0: + raise HTTPException(status_code=422, detail="Data length is zero") + + buf = bytearray(sz) + await get_data(buf, request) + + envelope = await asyncio.to_thread( + _submit_wait_solution, ctype, buf, accept + ) + return encode(envelope, accept, job_result=True) + + except (RequestValidationError, ValidationError) as e: + return encode(validation_exception_handler(e), accept) + + except HTTPException as e: + return encode(http_exception_handler(e), accept) + + except Exception as e: + return encode(exception_handler(e), accept) @app.post( diff --git a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py index 9a2fe7659f..da6572712b 100644 --- a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py +++ b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py @@ -204,6 +204,9 @@ def submit(self, problem, settings, enable_incumbents=None): def status(self, job_id): return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) + def wait(self, job_id, timeout=None): + return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) + def result(self, job_id, variable_names=None): status = self.jobs.get(job_id) if status in (FakeJobStatus.FAILED, FakeJobStatus.CANCELLED): @@ -254,6 +257,9 @@ def submit(self, data_model, settings=None): def status(self, job_id): return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) + def wait(self, job_id, timeout=0): + return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) + def result(self, job_id): status = self.jobs.get(job_id) if status in (FakeJobStatus.FAILED, FakeJobStatus.CANCELLED): @@ -858,10 +864,131 @@ def test_vrp_solution_after_sidecar_lost(proxy): assert "vehicle_data" in sol.json()["response"]["solver_response"] -def test_post_solution_and_sync_are_501(proxy): +def test_sync_cuopt_lp(proxy): + url, fake = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + body = res.json() + assert "reqId" not in body + assert body["response"]["solver_response"]["status"] == "Optimal" + # the managed path is stateless: no job or result is left behind + job_id = fake.submitted[0]["id"] + assert job_id in fake.deleted + import cuopt_server.proxy_webserver as pw + + with pw._jobs_lock: + assert job_id not in pw._jobs + + +def test_sync_cuopt_vrp(proxy): + url, fake = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_OptimizedRouting", + "data": _vrp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + body = res.json() + assert "reqId" not in body + solver_response = body["response"]["solver_response"] + assert solver_response["vehicle_data"]["veh-1"]["task_id"] == ["A"] + assert fake.routing.submitted[0]["id"] in fake.deleted + + +def test_sync_cuopt_validator_does_not_submit(proxy): + url, fake = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LPValidator", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + body = res.json() + assert "reqId" not in body + assert body["notes"] == ["Input is valid"] + assert fake.submitted == [] + + +def test_sync_cuopt_logs_nvcf_ids(proxy): + from cuopt_server.utils.logutil import get_ncaid, get_requestid + + url, _ = proxy + seen = {} + + def _capture(ctype, buf, accept): + seen["ncaid"] = get_ncaid() + seen["reqid"] = get_requestid() + return make_response({"solver_response": {"status": 0}}) + + import cuopt_server.proxy_webserver as pw + + original = pw._submit_wait_solution + pw._submit_wait_solution = _capture + try: + res = requests.post( + url + "/cuopt/cuopt", + headers={"NVCF-NCAID": "nca-1", "NVCF-REQID": "req-1"}, + json={"action": "cuOpt_LP", "data": _lp()}, + ) + finally: + pw._submit_wait_solution = original + assert res.status_code == 200, res.text + assert seen == {"ncaid": "nca-1", "reqid": "req-1"} + + +def test_sync_cuopt_rejects_zlib_content_type(proxy): + import zlib + + url, _ = proxy + payload = zlib.compress( + json.dumps({"action": "cuOpt_LP", "data": _lp()}).encode() + ) + res = requests.post( + url + "/cuopt/cuopt", + headers={"Content-Type": mime_zlib}, + data=payload, + ) + assert res.status_code == 415, res.text + + +def test_sync_cuopt_requires_wrapped_data(proxy): + url, _ = proxy + res = requests.post(url + "/cuopt/cuopt", json=_lp()) + assert res.status_code == 422, res.text + + +def test_sync_cuopt_rejects_null_data(proxy): + url, _ = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={"action": "cuOpt_LP", "data": None}, + ) + assert res.status_code == 422, res.text + assert "NVCF assets" in res.json()["error"] + + +def test_post_solution_and_warmstart_are_501(proxy): url, _ = proxy assert requests.post(url + "/cuopt/solution", json={}).status_code == 501 - assert requests.post(url + "/cuopt/cuopt", json={}).status_code == 501 + assert ( + requests.get( + url + f"/cuopt/solution/{uuid.uuid4()}/warmstart" + ).status_code + == 501 + ) assert requests.delete(url + "/cuopt/request/*").status_code == 501 assert ( requests.delete( From b31adb3ebb93e8d089a04d82b347a08f4434f40a Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Thu, 17 Sep 2026 16:03:23 -0400 Subject: [PATCH 2/5] proxy server log NVCF identifiers in sync endpoint --- .../cuopt_server/cuopt_server/cuopt_proxy.py | 27 ++++++- .../cuopt_server/proxy_webserver.py | 28 ++++++- .../tests/test_grpc_http_proxy.py | 78 +++++++++++++++++++ 3 files changed, 127 insertions(+), 6 deletions(-) diff --git a/python/cuopt_server/cuopt_server/cuopt_proxy.py b/python/cuopt_server/cuopt_server/cuopt_proxy.py index 709c5dd5c7..26a7f1e999 100644 --- a/python/cuopt_server/cuopt_server/cuopt_proxy.py +++ b/python/cuopt_server/cuopt_server/cuopt_proxy.py @@ -16,9 +16,14 @@ import cuopt_server.utils.settings as settings from cuopt_server._version import __version__ -from cuopt_server.utils.logutil import message_init - -log_fmt = "%(asctime)s.%(msecs)03d %(levelname)s %(message)s" +from cuopt_server.utils.logutil import ( + get_ncaid, + get_requestid, + get_solverid, + message_init, +) + +log_fmt = "%(ncaid)s%(requestid)s%(asctime)s.%(msecs)03d %(levelname)s %(message)s%(solverid)s" # noqa date_fmt = "%Y-%m-%d %H:%M:%S" @@ -165,6 +170,22 @@ def _configure_logging(args: argparse.Namespace) -> None: handlers=handlers, force=True, ) + log_factory = logging.getLogRecordFactory() + + def record_factory(*args, **kwargs): + record = log_factory(*args, **kwargs) + record.ncaid = get_ncaid() + record.requestid = get_requestid() + record.solverid = get_solverid() + if record.ncaid: + record.ncaid = f"NCA_ID={record.ncaid} " + if record.requestid: + record.requestid = f"NVCF_REQID={record.requestid} " + if record.solverid: + record.solverid = f" (GPU {record.solverid})" + return record + + logging.setLogRecordFactory(record_factory) def main(argv: Sequence[str] | None = None) -> None: diff --git a/python/cuopt_server/cuopt_server/proxy_webserver.py b/python/cuopt_server/cuopt_server/proxy_webserver.py index dcbca5f74e..728a5a40c0 100644 --- a/python/cuopt_server/cuopt_server/proxy_webserver.py +++ b/python/cuopt_server/cuopt_server/proxy_webserver.py @@ -92,7 +92,7 @@ validate_file_path, write_result_file, ) -from cuopt_server.utils.logutil import set_ncaid, set_requestid +from cuopt_server.utils.logutil import message, set_ncaid, set_requestid from cuopt_server.utils.routing.conversion import ( create_data_model as create_routing_data_model, create_solver as create_routing_solver, @@ -618,6 +618,7 @@ def _convert_and_submit( ) return job_id job_id = get_grpc_routing_client().submit(data_model, solver_settings) + logging.info(message(f"sent VRP job {job_id} to gRPC")) _store_job( job_id, { @@ -672,6 +673,7 @@ def _convert_and_submit( solver_settings, enable_incumbents=incumbents_enabled, ) + logging.info(message(f"sent LP job {job_id} to gRPC")) _store_job( job_id, { @@ -1109,27 +1111,47 @@ def _submit_wait_solution(ctype, buf, accept): "", None, ) + logging.info(message(f"submitted job {job_id}")) meta = _get_job(job_id) if meta is not None and meta.get("validation_only"): envelope = dict(meta["validation_result"]) envelope.pop("reqId", None) _pop_job(job_id) + logging.info(message(f"validation-only job {job_id} complete")) return envelope kind = None if meta is None else meta.get("kind") try: - get_grpc_client().wait(job_id) + logging.info(message(f"waiting for job {job_id}")) + try: + get_grpc_client().wait(job_id) + except Exception: + logging.error( + message(f"gRPC wait failed for job {job_id}"), + exc_info=True, + ) + raise status = get_grpc_client().status(job_id) + status_name = _status_name(status) + logging.info(message(f"gRPC status for job {job_id} is {status_name}")) if not _is_status(status, "COMPLETED"): + logging.error( + message( + f"gRPC job {job_id} finished unsuccessfully: {status_name}" + ) + ) raise HTTPException( status_code=409, - detail=f"job {_status_name(status).lower()}", + detail=f"job {status_name.lower()}", ) envelope, _warnings, _notes = _result_envelope(job_id, meta, kind) if envelope is None: + logging.error(message(f"gRPC job {job_id} returned no solution")) raise HTTPException( status_code=500, detail="solver returned no solution" ) + logging.info(message(f"received result for job {job_id}")) + logging.info({"cuopt_complete": status_name}) return envelope finally: try: diff --git a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py index da6572712b..2b5e2c8ec7 100644 --- a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py +++ b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 import json +import logging import socket import threading import time @@ -887,6 +888,27 @@ def test_sync_cuopt_lp(proxy): assert job_id not in pw._jobs +def test_sync_cuopt_logs_job_lifecycle(proxy, caplog): + url, fake = proxy + with caplog.at_level(logging.INFO): + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + job_id = fake.submitted[0]["id"] + text = caplog.text + assert f"sent LP job {job_id} to gRPC" in text + assert f"submitted job {job_id}" in text + assert f"waiting for job {job_id}" in text + assert f"gRPC status for job {job_id} is COMPLETED" in text + assert f"received result for job {job_id}" in text + + def test_sync_cuopt_vrp(proxy): url, fake = proxy res = requests.post( @@ -949,6 +971,62 @@ def _capture(ctype, buf, accept): assert seen == {"ncaid": "nca-1", "reqid": "req-1"} +def test_sync_cuopt_log_records_include_nvcf_ids(proxy, caplog): + from cuopt_server.utils.logutil import ( + get_ncaid, + get_requestid, + get_solverid, + ) + + url, fake = proxy + previous = logging.getLogRecordFactory() + + def record_factory(*args, **kwargs): + record = previous(*args, **kwargs) + record.ncaid = get_ncaid() + record.requestid = get_requestid() + record.solverid = get_solverid() + if record.ncaid: + record.ncaid = f"NCA_ID={record.ncaid} " + if record.requestid: + record.requestid = f"NVCF_REQID={record.requestid} " + if record.solverid: + record.solverid = f" (GPU {record.solverid})" + return record + + logging.setLogRecordFactory(record_factory) + try: + with caplog.at_level(logging.INFO): + res = requests.post( + url + "/cuopt/cuopt", + headers={"NVCF-NCAID": "nca-1", "NVCF-REQID": "req-1"}, + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + finally: + logging.setLogRecordFactory(previous) + assert res.status_code == 200, res.text + job_id = fake.submitted[0]["id"] + markers = ( + f"sent LP job {job_id} to gRPC", + f"submitted job {job_id}", + f"waiting for job {job_id}", + f"received result for job {job_id}", + ) + lifecycle = [ + rec + for rec in caplog.records + if any(marker in rec.getMessage() for marker in markers) + ] + assert lifecycle, caplog.text + for rec in lifecycle: + assert rec.ncaid == "NCA_ID=nca-1 " + assert rec.requestid == "NVCF_REQID=req-1 " + + def test_sync_cuopt_rejects_zlib_content_type(proxy): import zlib From c3c5eb698f984d64269da49d9335d80c06a25ca4 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Fri, 18 Sep 2026 12:25:32 -0400 Subject: [PATCH 3/5] proxy server apply feedback for cuopt/cuopt endpoint --- .../cuopt_server/proxy_webserver.py | 73 ++++++++++++++----- .../tests/test_grpc_http_proxy.py | 65 ++++++++++++++--- 2 files changed, 109 insertions(+), 29 deletions(-) diff --git a/python/cuopt_server/cuopt_server/proxy_webserver.py b/python/cuopt_server/cuopt_server/proxy_webserver.py index 728a5a40c0..5a11866579 100644 --- a/python/cuopt_server/cuopt_server/proxy_webserver.py +++ b/python/cuopt_server/cuopt_server/proxy_webserver.py @@ -4,6 +4,7 @@ """FastAPI app for the gRPC-backed HTTP proxy (LP/MILP/VRP).""" import asyncio +import contextlib import logging import os import threading @@ -122,6 +123,13 @@ _incumbent_locks = {} _max_request_size = 1024 * 1024 * 1024 +# Managed POST /cuopt/cuopt polls job status from the event loop rather than +# calling Client.wait, which would hold a thread-pool worker for the whole +# solve and starve other requests, including the health check. Back off up to +# _STATUS_POLL_MAX so long solves do not poll thousands of times. +_STATUS_POLL_MIN = 0.05 +_STATUS_POLL_MAX = 1.0 + _ROUTING_KEYS = { "cost_matrix_data", "task_data", @@ -567,6 +575,7 @@ def _deserialize_convert_submit( solver_logs, accept, result_file, + warmstart_id, initial_ids, ) @@ -579,6 +588,7 @@ def _convert_and_submit( solver_logs, accept, result_file, + warmstart_id, initial_ids, ): """Convert a decoded problem body and submit it over gRPC.""" @@ -1073,13 +1083,8 @@ def getrequest( return encode(exception_handler(e), accept) -def _submit_wait_solution(ctype, buf, accept): - """Submit, wait, and return a solution for POST /cuopt/cuopt. - - The managed endpoint is stateless: the gRPC job and the proxy metadata - are released before returning, so there is nothing left to poll or - delete afterwards. - """ +def _submit_managed_job(ctype, buf, accept): + """Validate and submit a POST /cuopt/cuopt body, returning the job id.""" body = deserialize(ctype, buf) try: wrapper = cuoptDataInternal.parse_obj(body) @@ -1109,9 +1114,41 @@ def _submit_wait_solution(ctype, buf, accept): False, accept, "", + "", None, ) logging.info(message(f"submitted job {job_id}")) + return job_id + + +async def _poll_job_status(job_id): + """Return the terminal gRPC status, yielding the loop while job runs.""" + client = get_grpc_client() + interval = _STATUS_POLL_MIN + while True: + status = await asyncio.to_thread(client.status, job_id) + if not _is_status(status, "QUEUED", "PROCESSING"): + return status + await asyncio.sleep(interval) + interval = min(interval * 2, _STATUS_POLL_MAX) + + +def _release_managed_job(job_id): + try: + get_grpc_client().delete(job_id) + except Exception: + logging.warning(f"could not delete job {job_id}", exc_info=True) + _pop_job(job_id) + + +async def _submit_wait_solution(ctype, buf, accept): + """Submit, wait, and return a solution for POST /cuopt/cuopt. + + The managed endpoint is stateless: the gRPC job and the proxy metadata + are released before returning, so there is nothing left to poll or + delete afterwards. + """ + job_id = await asyncio.to_thread(_submit_managed_job, ctype, buf, accept) meta = _get_job(job_id) if meta is not None and meta.get("validation_only"): envelope = dict(meta["validation_result"]) @@ -1124,14 +1161,13 @@ def _submit_wait_solution(ctype, buf, accept): try: logging.info(message(f"waiting for job {job_id}")) try: - get_grpc_client().wait(job_id) + status = await _poll_job_status(job_id) except Exception: logging.error( message(f"gRPC wait failed for job {job_id}"), exc_info=True, ) raise - status = get_grpc_client().status(job_id) status_name = _status_name(status) logging.info(message(f"gRPC status for job {job_id} is {status_name}")) if not _is_status(status, "COMPLETED"): @@ -1144,7 +1180,9 @@ def _submit_wait_solution(ctype, buf, accept): status_code=409, detail=f"job {status_name.lower()}", ) - envelope, _warnings, _notes = _result_envelope(job_id, meta, kind) + envelope, _warnings, _notes = await asyncio.to_thread( + _result_envelope, job_id, meta, kind + ) if envelope is None: logging.error(message(f"gRPC job {job_id} returned no solution")) raise HTTPException( @@ -1154,11 +1192,12 @@ def _submit_wait_solution(ctype, buf, accept): logging.info({"cuopt_complete": status_name}) return envelope finally: - try: - get_grpc_client().delete(job_id) - except Exception: - logging.warning(f"could not delete job {job_id}", exc_info=True) - _pop_job(job_id) + # shielded so a client disconnect mid-solve still frees the job + cleanup = asyncio.ensure_future( + asyncio.to_thread(_release_managed_job, job_id) + ) + with contextlib.suppress(asyncio.CancelledError): + await asyncio.shield(cleanup) @app.post( @@ -1237,9 +1276,7 @@ async def cuopt( buf = bytearray(sz) await get_data(buf, request) - envelope = await asyncio.to_thread( - _submit_wait_solution, ctype, buf, accept - ) + envelope = await _submit_wait_solution(ctype, buf, accept) return encode(envelope, accept, job_result=True) except (RequestValidationError, ValidationError) as e: diff --git a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py index 2b5e2c8ec7..dc714375e2 100644 --- a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py +++ b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py @@ -7,6 +7,7 @@ import threading import time import uuid +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import pytest @@ -182,6 +183,9 @@ def __init__(self): self.submitted = [] self.cancelled = [] self.deleted = [] + self.waits = [] + self.status_calls = [] + self.pending_statuses = 0 self._incumbents = {} self._logs = {} @@ -203,9 +207,14 @@ def submit(self, problem, settings, enable_incumbents=None): return job_id def status(self, job_id): + self.status_calls.append(job_id) + if self.pending_statuses > 0: + self.pending_statuses -= 1 + return FakeJobStatus.PROCESSING return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) def wait(self, job_id, timeout=None): + self.waits.append(timeout) return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) def result(self, job_id, variable_names=None): @@ -258,9 +267,6 @@ def submit(self, data_model, settings=None): def status(self, job_id): return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) - def wait(self, job_id, timeout=0): - return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) - def result(self, job_id): status = self.jobs.get(job_id) if status in (FakeJobStatus.FAILED, FakeJobStatus.CANCELLED): @@ -882,6 +888,8 @@ def test_sync_cuopt_lp(proxy): # the managed path is stateless: no job or result is left behind job_id = fake.submitted[0]["id"] assert job_id in fake.deleted + # the endpoint polls status instead of blocking a thread in Client.wait + assert fake.waits == [] import cuopt_server.proxy_webserver as pw with pw._jobs_lock: @@ -950,7 +958,7 @@ def test_sync_cuopt_logs_nvcf_ids(proxy): url, _ = proxy seen = {} - def _capture(ctype, buf, accept): + async def _capture(ctype, buf, accept): seen["ncaid"] = get_ncaid() seen["reqid"] = get_requestid() return make_response({"solver_response": {"status": 0}}) @@ -1048,6 +1056,47 @@ def test_sync_cuopt_requires_wrapped_data(proxy): assert res.status_code == 422, res.text +def test_sync_cuopt_polls_until_job_is_done(proxy): + url, fake = proxy + fake.pending_statuses = 3 + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + assert res.json()["response"]["solver_response"]["status"] == "Optimal" + job_id = fake.submitted[0]["id"] + assert fake.status_calls.count(job_id) >= 4 + assert fake.waits == [] + assert job_id in fake.deleted + + +def test_sync_cuopt_health_is_served_during_solve(proxy): + """A pending solve must not hold the thread pool that health needs.""" + url, fake = proxy + fake.pending_statuses = 5 + + with ThreadPoolExecutor(max_workers=1) as pool: + solve = pool.submit( + requests.post, + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + while fake.pending_statuses > 3: + time.sleep(0.01) + health = requests.get(url + "/cuopt/health", timeout=5) + assert health.status_code == 200, health.text + assert solve.result().status_code == 200 + + def test_sync_cuopt_rejects_null_data(proxy): url, _ = proxy res = requests.post( @@ -1058,15 +1107,9 @@ def test_sync_cuopt_rejects_null_data(proxy): assert "NVCF assets" in res.json()["error"] -def test_post_solution_and_warmstart_are_501(proxy): +def test_post_solution_and_wildcards_are_501(proxy): url, _ = proxy assert requests.post(url + "/cuopt/solution", json={}).status_code == 501 - assert ( - requests.get( - url + f"/cuopt/solution/{uuid.uuid4()}/warmstart" - ).status_code - == 501 - ) assert requests.delete(url + "/cuopt/request/*").status_code == 501 assert ( requests.delete( From 6ff2f0899429dc4ffdaa6c6067f98ecc9a9ebeb7 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Fri, 18 Sep 2026 13:58:02 -0400 Subject: [PATCH 4/5] grpc server add timeouts to health-during-solve test --- .../cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py index dc714375e2..1e3a27f687 100644 --- a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py +++ b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py @@ -1090,11 +1090,14 @@ def test_sync_cuopt_health_is_served_during_solve(proxy): "client_version": "custom", }, ) + deadline = time.monotonic() + 5 while fake.pending_statuses > 3: + if time.monotonic() > deadline: + pytest.fail("solve never started polling") time.sleep(0.01) health = requests.get(url + "/cuopt/health", timeout=5) assert health.status_code == 200, health.text - assert solve.result().status_code == 200 + assert solve.result(timeout=10).status_code == 200 def test_sync_cuopt_rejects_null_data(proxy): From a091798d34eabdeefef0294a9771665bb01ebeee Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Mon, 21 Sep 2026 17:03:43 -0400 Subject: [PATCH 5/5] proxy server sync endpoint apply feedback * missed warmstart cache write in refactor * timeout adjustment on test --- .../cuopt_server/proxy_webserver.py | 12 ++++++++-- .../tests/test_grpc_http_proxy.py | 24 +++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/python/cuopt_server/cuopt_server/proxy_webserver.py b/python/cuopt_server/cuopt_server/proxy_webserver.py index 5a11866579..0b24aac473 100644 --- a/python/cuopt_server/cuopt_server/proxy_webserver.py +++ b/python/cuopt_server/cuopt_server/proxy_webserver.py @@ -945,10 +945,14 @@ def deleterequest( return encode(exception_handler(e), accept) -def _result_envelope(job_id, meta, kind, req_id=""): +def _result_envelope(job_id, meta, kind, req_id="", cache_warmstart=False): """Build the legacy solution envelope for a finished job. Returns ``(None, [], [])`` when the result is not available yet. + + ``cache_warmstart`` populates the warmstart cache from an LP solution so a + later GET of the warmstart route does not have to refetch and reparse the + result. Callers that release the job before returning leave it off. """ result_kind, sol = _result_for_job(job_id, meta, kind) if sol is None: @@ -966,6 +970,8 @@ def _result_envelope(job_id, meta, kind, req_id=""): notes.append(sol.get("status_message") or "") notes = [n for n in notes if n] else: + if cache_warmstart: + _store_warmstart(job_id, _warmstart_dict_from_sol(sol)) inner = solution_to_http(sol, include_warmstart=False) try: notes.append(sol.get_termination_reason()) @@ -1034,7 +1040,9 @@ def getsolution( status_code=409, detail=f"job {id} {_status_name(status).lower()}", ) - envelope, warnings, notes = _result_envelope(id, meta, kind, req_id=id) + envelope, warnings, notes = _result_envelope( + id, meta, kind, req_id=id, cache_warmstart=True + ) if envelope is None: return encode({"reqId": id}, accept) resultdir, maxresult, mode = settings.get_result_dir() diff --git a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py index 1e3a27f687..0e9440d6f0 100644 --- a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py +++ b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py @@ -185,6 +185,7 @@ def __init__(self): self.deleted = [] self.waits = [] self.status_calls = [] + self.result_calls = [] self.pending_statuses = 0 self._incumbents = {} self._logs = {} @@ -218,6 +219,7 @@ def wait(self, job_id, timeout=None): return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) def result(self, job_id, variable_names=None): + self.result_calls.append(job_id) status = self.jobs.get(job_id) if status in (FakeJobStatus.FAILED, FakeJobStatus.CANCELLED): raise RuntimeError(f"job {status.name.lower()}") @@ -544,6 +546,27 @@ def test_warmstart_get_and_reuse(proxy): assert list(ws.current_primal_solution) == [0.1, 0.2] +def test_getsolution_caches_warmstart(proxy): + import cuopt_server.proxy_webserver as pw + + url, fake = proxy + req_id = requests.post( + url + "/cuopt/request", + headers={"CLIENT-VERSION": "custom"}, + json=_lp(), + ).json()["reqId"] + + assert requests.get(url + f"/cuopt/solution/{req_id}").status_code == 200 + assert pw._cached_warmstart(req_id) is not None + + # Cached by the solution GET, so the warmstart route serves it without + # refetching the result over gRPC. + calls = len(fake.result_calls) + warm = requests.get(url + f"/cuopt/solution/{req_id}/warmstart") + assert warm.status_code == 200, warm.text + assert fake.result_calls[calls:] == [] + + def test_warmstart_missing_id_is_404(proxy): url, _ = proxy missing = str(uuid.uuid4()) @@ -1089,6 +1112,7 @@ def test_sync_cuopt_health_is_served_during_solve(proxy): "data": _lp(), "client_version": "custom", }, + timeout=8, ) deadline = time.monotonic() + 5 while fake.pending_statuses > 3: