admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-27 21:05:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix blocking async server handlers (#618)

Browse Source 
Closes #219. Replace synchronous httpx calls in async SendBlue and model-management handlers with awaited httpx.AsyncClient (context-managed close); run Whisper transcription and engine.list_models via asyncio.to_thread so they don't block the event loop; and harden TelemetryStore/aggregator SQLite for concurrency (WAL, synchronous=NORMAL, busy_timeout=5000, plus a write-serializing lock on the shared connection). Adds async-usage assertions and a real 8-thread concurrent-write test. Related: #570 (async httpx, different issue #559).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Elliot Slusky
2026-06-30 17:37:20 -07:00
committed by GitHub
co-authored by Claude Opus 4.8
parent c686517cc7
commit d865b4bed4
10 changed files with 308 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/server/agent_manager_routes.py
+40 -36
View File
@@ -2263,14 +2263,14 @@ def create_agent_manager_router(
import httpx
try:
resp = httpx.get(
"https://api.sendblue.co/api/lines",
headers={
"sb-api-key-id": api_key_id,
"sb-api-secret-key": api_secret_key,
},
timeout=15.0,
)
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.get(
"https://api.sendblue.co/api/lines",
headers={
"sb-api-key-id": api_key_id,
"sb-api-secret-key": api_secret_key,
},
)
if resp.status_code == 401:
raise HTTPException(
status_code=401,
@@ -2290,12 +2290,16 @@ def create_agent_manager_router(
)
numbers = []
for line in lines:
num = (
line.get("number")
or line.get("phone_number")
or line.get("from_number")
or (line if isinstance(line, str) else "")
)
if isinstance(line, str):
num = line
elif isinstance(line, dict):
num = (
line.get("number")
or line.get("phone_number")
or line.get("from_number")
)
else:
num = None
if num:
numbers.append(num)
return {
@@ -2327,18 +2331,18 @@ def create_agent_manager_router(
import httpx
try:
resp = httpx.post(
"https://api.sendblue.co/api/account/webhooks",
headers={
"sb-api-key-id": api_key_id,
"sb-api-secret-key": api_secret_key,
"Content-Type": "application/json",
},
json={
"receive": webhook_url,
},
timeout=15.0,
)
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.post(
"https://api.sendblue.co/api/account/webhooks",
headers={
"sb-api-key-id": api_key_id,
"sb-api-secret-key": api_secret_key,
"Content-Type": "application/json",
},
json={
"receive": webhook_url,
},
)
return {
"registered": resp.status_code < 300,
"status": resp.status_code,
@@ -2378,16 +2382,16 @@ def create_agent_manager_router(
if from_number:
payload["from_number"] = from_number
resp = httpx.post(
"https://api.sendblue.co/api/send-message",
headers={
"sb-api-key-id": api_key_id,
"sb-api-secret-key": api_secret_key,
"Content-Type": "application/json",
},
json=payload,
timeout=15.0,
)
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.post(
"https://api.sendblue.co/api/send-message",
headers={
"sb-api-key-id": api_key_id,
"sb-api-secret-key": api_secret_key,
"Content-Type": "application/json",
},
json=payload,
)
return {
"sent": resp.status_code < 300,
"status": resp.status_code,
src/openjarvis/server/api_routes.py
+7 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import inspect
import json
import logging
@@ -894,7 +895,12 @@ async def transcribe_speech(request: Request):
ext = filename.rsplit(".", 1)[-1] if "." in filename else "wav"
try:
result = backend.transcribe(audio_bytes, format=ext, language=language or None)
result = await asyncio.to_thread(
backend.transcribe,
audio_bytes,
format=ext,
language=language or None,
)
except Exception as exc:
logger.exception("Speech transcription failed")
raise HTTPException(
src/openjarvis/server/routes.py
+13 -16
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import logging
import uuid
from typing import Any
@@ -835,7 +836,7 @@ async def list_models(request: Request) -> ModelListResponse:
# Filter out any cloud model IDs that may appear via MultiEngine.
# Fall back to direct Ollama query only when the engine returns nothing.
engine = request.app.state.engine
all_ids = engine.list_models()
all_ids = await asyncio.to_thread(engine.list_models)
model_ids = [m for m in all_ids if not is_cloud_model(m)]
if not model_ids:
model_ids = await list_local_models()
@@ -865,12 +866,12 @@ async def pull_model(request: Request):
import httpx as _httpx
host = getattr(engine, "_host", "http://localhost:11434")
client = _httpx.Client(base_url=host, timeout=600.0)
try:
resp = client.post(
"/api/pull",
json={"name": model_name, "stream": False},
)
async with _httpx.AsyncClient(base_url=host, timeout=600.0) as client:
resp = await client.post(
"/api/pull",
json={"name": model_name, "stream": False},
)
resp.raise_for_status()
except (_httpx.ConnectError, _httpx.TimeoutException) as exc:
raise HTTPException(status_code=502, detail=f"Ollama unreachable: {exc}")
@@ -879,8 +880,6 @@ async def pull_model(request: Request):
status_code=exc.response.status_code,
detail=f"Ollama error: {exc.response.text[:300]}",
)
finally:
client.close()
return {"status": "ok", "model": model_name}
@@ -896,13 +895,13 @@ async def delete_model(model_name: str, request: Request):
import httpx as _httpx
host = getattr(engine, "_host", "http://localhost:11434")
client = _httpx.Client(base_url=host, timeout=30.0)
try:
resp = client.request(
"DELETE",
"/api/delete",
json={"name": model_name},
)
async with _httpx.AsyncClient(base_url=host, timeout=30.0) as client:
resp = await client.request(
"DELETE",
"/api/delete",
json={"name": model_name},
)
resp.raise_for_status()
except (_httpx.ConnectError, _httpx.TimeoutException) as exc:
raise HTTPException(status_code=502, detail=f"Ollama unreachable: {exc}")
@@ -911,8 +910,6 @@ async def delete_model(model_name: str, request: Request):
status_code=exc.response.status_code,
detail=f"Ollama error: {exc.response.text[:300]}",
)
finally:
client.close()
return {"status": "deleted", "model": model_name}
src/openjarvis/telemetry/aggregator.py
+3
View File
@@ -90,6 +90,9 @@ class TelemetryAggregator:
def __init__(self, db_path: str | Path) -> None:
self._db_path = str(db_path)
self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("PRAGMA synchronous=NORMAL")
self._conn.execute("PRAGMA busy_timeout=5000")
self._conn.row_factory = sqlite3.Row
def _time_filter(
src/openjarvis/telemetry/store.py
+71 -64
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import json
import logging
import sqlite3
import threading
import time
from pathlib import Path
from typing import Any
@@ -148,6 +149,10 @@ class TelemetryStore:
def __init__(self, db_path: str | Path) -> None:
self._db_path = str(db_path)
self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
self._lock = threading.Lock()
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("PRAGMA synchronous=NORMAL")
self._conn.execute("PRAGMA busy_timeout=5000")
self._conn.execute(_CREATE_TABLE)
self._conn.execute(_CREATE_MINING_STATS_TABLE)
self._conn.commit()
@@ -166,53 +171,54 @@ class TelemetryStore:
def record(self, rec: TelemetryRecord) -> None:
"""Persist a single telemetry record."""
self._conn.execute(
_INSERT,
(
rec.timestamp,
rec.model_id,
rec.engine,
rec.agent,
rec.prompt_tokens,
rec.prompt_tokens_evaluated,
rec.completion_tokens,
rec.total_tokens,
rec.latency_seconds,
rec.ttft,
rec.cost_usd,
rec.energy_joules,
rec.power_watts,
rec.gpu_utilization_pct,
rec.gpu_memory_used_gb,
rec.gpu_temperature_c,
rec.throughput_tok_per_sec,
rec.prefill_latency_seconds,
rec.decode_latency_seconds,
rec.energy_method,
rec.energy_vendor,
rec.batch_id,
1 if rec.is_warmup else 0,
rec.cpu_energy_joules,
rec.gpu_energy_joules,
rec.dram_energy_joules,
rec.tokens_per_joule,
rec.energy_per_output_token_joules,
rec.throughput_per_watt,
rec.prefill_energy_joules,
rec.decode_energy_joules,
rec.mean_itl_ms,
rec.median_itl_ms,
rec.p90_itl_ms,
rec.p95_itl_ms,
rec.p99_itl_ms,
rec.std_itl_ms,
1 if rec.is_streaming else 0,
rec.token_counting_version,
rec.mining_session_id,
json.dumps(rec.metadata),
),
)
self._conn.commit()
with self._lock:
self._conn.execute(
_INSERT,
(
rec.timestamp,
rec.model_id,
rec.engine,
rec.agent,
rec.prompt_tokens,
rec.prompt_tokens_evaluated,
rec.completion_tokens,
rec.total_tokens,
rec.latency_seconds,
rec.ttft,
rec.cost_usd,
rec.energy_joules,
rec.power_watts,
rec.gpu_utilization_pct,
rec.gpu_memory_used_gb,
rec.gpu_temperature_c,
rec.throughput_tok_per_sec,
rec.prefill_latency_seconds,
rec.decode_latency_seconds,
rec.energy_method,
rec.energy_vendor,
rec.batch_id,
1 if rec.is_warmup else 0,
rec.cpu_energy_joules,
rec.gpu_energy_joules,
rec.dram_energy_joules,
rec.tokens_per_joule,
rec.energy_per_output_token_joules,
rec.throughput_per_watt,
rec.prefill_energy_joules,
rec.decode_energy_joules,
rec.mean_itl_ms,
rec.median_itl_ms,
rec.p90_itl_ms,
rec.p95_itl_ms,
rec.p99_itl_ms,
rec.std_itl_ms,
1 if rec.is_streaming else 0,
rec.token_counting_version,
rec.mining_session_id,
json.dumps(rec.metadata),
),
)
self._conn.commit()
def record_mining_stats(self, stats: Any) -> None:
"""Persist one mining stats snapshot.
@@ -220,28 +226,29 @@ class TelemetryStore:
``stats`` is duck-typed to keep telemetry usable without importing the
optional mining package at module import time.
"""
self._conn.execute(
"""\
with self._lock:
self._conn.execute(
"""\
INSERT INTO mining_stats (
recorded_at, provider_id, shares_submitted, shares_accepted, blocks_found,
hashrate, uptime_seconds, last_share_at, last_error, payout_target, fees_owed
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
time.time(),
stats.provider_id,
stats.shares_submitted,
stats.shares_accepted,
stats.blocks_found,
stats.hashrate,
stats.uptime_seconds,
stats.last_share_at,
stats.last_error,
stats.payout_target,
stats.fees_owed,
),
)
self._conn.commit()
(
time.time(),
stats.provider_id,
stats.shares_submitted,
stats.shares_accepted,
stats.blocks_found,
stats.hashrate,
stats.uptime_seconds,
stats.last_share_at,
stats.last_error,
stats.payout_target,
stats.fees_owed,
),
)
self._conn.commit()
def list_recent(self, limit: int = 50) -> list[dict[str, Any]]:
"""Return recent telemetry rows as dictionaries."""
tests/server/test_agent_manager_routes.py
+66 -1
View File
@@ -6,7 +6,7 @@ import json
import tempfile
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -48,6 +48,71 @@ class TestAgentManagerRoutes:
assert resp.status_code == 200
assert resp.json()["agents"] == []
def test_sendblue_verify_uses_async_http_client(self, client):
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = [
"+15551234567",
{"phone_number": "+15557654321"},
None,
]
with patch("httpx.AsyncClient") as mock_client_cls:
instance = mock_client_cls.return_value.__aenter__.return_value
instance.get = AsyncMock(return_value=mock_resp)
resp = client.post(
"/v1/channels/sendblue/verify",
json={"api_key_id": "key-id", "api_secret_key": "secret"},
)
assert resp.status_code == 200
assert resp.json()["valid"] is True
assert resp.json()["numbers"] == ["+15551234567", "+15557654321"]
instance.get.assert_awaited_once()
def test_sendblue_register_webhook_uses_async_http_client(self, client):
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"ok": True}
with patch("httpx.AsyncClient") as mock_client_cls:
instance = mock_client_cls.return_value.__aenter__.return_value
instance.post = AsyncMock(return_value=mock_resp)
resp = client.post(
"/v1/channels/sendblue/register-webhook",
json={
"api_key_id": "key-id",
"api_secret_key": "secret",
"webhook_url": "https://example.com/webhooks/sendblue",
},
)
assert resp.status_code == 200
assert resp.json()["registered"] is True
instance.post.assert_awaited_once()
def test_sendblue_test_message_uses_async_http_client(self, client):
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"ok": True}
with patch("httpx.AsyncClient") as mock_client_cls:
instance = mock_client_cls.return_value.__aenter__.return_value
instance.post = AsyncMock(return_value=mock_resp)
resp = client.post(
"/v1/channels/sendblue/test",
json={
"api_key_id": "key-id",
"api_secret_key": "secret",
"from_number": "+15550000000",
"to_number": "+15551234567",
},
)
assert resp.status_code == 200
assert resp.json()["sent"] is True
instance.post.assert_awaited_once()
def test_create_agent(self, client):
resp = client.post(
"/v1/managed-agents",
tests/server/test_model_management.py
+35 -13
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -75,10 +75,9 @@ class TestModelPull:
mock_resp.status_code = 200
mock_resp.raise_for_status = MagicMock()
with patch("httpx.Client") as MockClient:
instance = MockClient.return_value
instance.post.return_value = mock_resp
instance.close = MagicMock()
with patch("httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(return_value=mock_resp)
resp = client.post("/v1/models/pull", json={"model": "qwen3.5:4b"})
@@ -86,6 +85,10 @@ class TestModelPull:
data = resp.json()
assert data["status"] == "ok"
assert data["model"] == "qwen3.5:4b"
instance.post.assert_awaited_once_with(
"/api/pull",
json={"name": "qwen3.5:4b", "stream": False},
)
def test_pull_ollama_unreachable(self):
engine = _make_ollama_engine()
@@ -93,10 +96,9 @@ class TestModelPull:
import httpx
with patch("httpx.Client") as MockClient:
instance = MockClient.return_value
instance.post.side_effect = httpx.ConnectError("refused")
instance.close = MagicMock()
with patch("httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(side_effect=httpx.ConnectError("refused"))
resp = client.post("/v1/models/pull", json={"model": "foo"})
@@ -123,10 +125,9 @@ class TestModelDelete:
mock_resp.status_code = 200
mock_resp.raise_for_status = MagicMock()
with patch("httpx.Client") as MockClient:
instance = MockClient.return_value
instance.request.return_value = mock_resp
instance.close = MagicMock()
with patch("httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.request = AsyncMock(return_value=mock_resp)
resp = client.delete("/v1/models/qwen3:0.6b")
@@ -134,6 +135,11 @@ class TestModelDelete:
data = resp.json()
assert data["status"] == "deleted"
assert data["model"] == "qwen3:0.6b"
instance.request.assert_awaited_once_with(
"DELETE",
"/api/delete",
json={"name": "qwen3:0.6b"},
)
# ---------------------------------------------------------------------------
@@ -268,3 +274,19 @@ class TestModelsEndpointExtended:
assert resp.status_code == 200
# The endpoint returns whatever list_models() gives
assert resp.json()["object"] == "list"
def test_models_list_offloads_engine_list_models(self):
engine = _make_engine(models=["qwen3.5:4b"])
app = create_app(engine, "qwen3.5:4b")
client = TestClient(app)
with patch(
"openjarvis.server.routes.asyncio.to_thread",
new_callable=AsyncMock,
) as mock_to_thread:
mock_to_thread.return_value = ["qwen3.5:4b"]
resp = client.get("/v1/models")
assert resp.status_code == 200
assert [m["id"] for m in resp.json()["data"]] == ["qwen3.5:4b"]
mock_to_thread.assert_awaited_once_with(engine.list_models)
tests/server/test_speech_routes.py
+28 -1
View File
@@ -1,6 +1,6 @@
"""Tests for speech API endpoints."""
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -56,6 +56,33 @@ def test_transcribe_endpoint(client, mock_speech_backend):
assert data["duration_seconds"] == 1.5
def test_transcribe_endpoint_offloads_backend_work(client, mock_speech_backend):
expected = TranscriptionResult(
text="Offloaded",
language="en",
confidence=0.9,
duration_seconds=1.0,
segments=[],
)
with patch(
"openjarvis.server.api_routes.asyncio.to_thread",
new_callable=AsyncMock,
) as mock_to_thread:
mock_to_thread.return_value = expected
response = client.post(
"/v1/speech/transcribe",
files={"file": ("test.wav", b"fake audio data", "audio/wav")},
)
assert response.status_code == 200
mock_to_thread.assert_awaited_once()
args, kwargs = mock_to_thread.await_args
assert args == (mock_speech_backend.transcribe, b"fake audio data")
assert kwargs == {"format": "wav", "language": None}
assert response.json()["text"] == "Offloaded"
def test_transcribe_endpoint_surfaces_backend_error(client, mock_speech_backend):
mock_speech_backend.transcribe.side_effect = RuntimeError("missing cublas64_12.dll")
tests/telemetry/test_aggregator.py
+15
View File
@@ -49,6 +49,21 @@ def _setup(tmp_path: Path, records: list[TelemetryRecord] | None = None):
class TestTelemetryAggregator:
def test_uses_wal_with_normal_synchronous_and_busy_timeout(
self,
tmp_path: Path,
) -> None:
agg = _setup(tmp_path)
journal_mode = agg._conn.execute("PRAGMA journal_mode").fetchone()[0]
synchronous = agg._conn.execute("PRAGMA synchronous").fetchone()[0]
busy_timeout = agg._conn.execute("PRAGMA busy_timeout").fetchone()[0]
assert journal_mode.lower() == "wal"
assert synchronous == 1
assert busy_timeout == 5000
agg.close()
def test_empty_db_summary(self, tmp_path: Path) -> None:
agg = _setup(tmp_path)
s = agg.summary()
tests/telemetry/test_store.py
+30
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from openjarvis.core.events import EventBus, EventType
@@ -17,6 +18,35 @@ class TestTelemetryStore:
assert rows == []
store.close()
def test_uses_wal_with_normal_synchronous(self, tmp_path: Path) -> None:
store = TelemetryStore(tmp_path / "test.db")
journal_mode = store._conn.execute("PRAGMA journal_mode").fetchone()[0]
synchronous = store._conn.execute("PRAGMA synchronous").fetchone()[0]
busy_timeout = store._conn.execute("PRAGMA busy_timeout").fetchone()[0]
assert journal_mode.lower() == "wal"
assert synchronous == 1
assert busy_timeout == 5000
store.close()
def test_concurrent_record_writes_are_serialized(self, tmp_path: Path) -> None:
store = TelemetryStore(tmp_path / "test.db")
def write_one(i: int) -> None:
store.record(
TelemetryRecord(
timestamp=time.time(),
model_id=f"model-{i}",
engine="test",
)
)
with ThreadPoolExecutor(max_workers=8) as pool:
list(pool.map(write_one, range(32)))
assert len(store._fetchall()) == 32
store.close()
def test_record_values(self, tmp_path: Path) -> None:
store = TelemetryStore(tmp_path / "test.db")
rec = TelemetryRecord(