swadhinbiswas 🚀 release: 2.0.8 — write-path fix (ADR-080), fork-safe GIL pool (ADR-081), HTTP/3 (ADR-082), Security DI fix, docs honesty pass · 8/6/2026
731fffe
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
"""Server-path concurrent-write durability test (P2.2 regression guard).
Uses the async test client with many concurrent coroutines issuing INSERTs,
then asserts every row persisted. This exercises the same `app.db.query`
write path a real server handler uses, under genuine concurrency — without
the single-worker GIL-pool TCP stall that a raw `oha -c 50` against the
blocking server would hit.
"""
import os
import tempfile
import pytest
from justapi import JustAPIApp
from justapi.testing import AsyncTestClient
@pytest.mark.asyncio
async def test_concurrent_writes_via_test_client():
tmpdir = tempfile.mkdtemp()
db_path = os.path.join(tmpdir, "svc_concurrent.db")
open(db_path, "w").close()
try:
app = JustAPIApp()
app.set_database(f"sqlite://{db_path}")
@app.post("/items")
async def create(request):
body = request.json()
# Fire-and-forget style: the write must persist under concurrency.
app.db.query(
"INSERT INTO items(name, qty) VALUES (?, ?)",
[body["name"], body["qty"]],
)
return {"ok": True}
@app.get("/items/{id}")
async def read(request):
row = app.db.query(
"SELECT * FROM items WHERE id = ?",
[int(request["path_params"]["id"])],
)
return row[0] if row else {"status": 404}
async with AsyncTestClient(app, database=f"sqlite://{db_path}") as c:
# Create the table on the same file pool the handlers use.
app.db.query(
"CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT, qty INTEGER)"
)
import asyncio
async def insert_one(v):
return await c.post(
"/items", f'{{"name":"w{v}","qty":{v}}}'.encode()
)
results = await asyncio.gather(
*(insert_one(v) for v in range(100)),
return_exceptions=True,
)
errors = [r for r in results if isinstance(r, Exception)]
assert not errors, f"{len(errors)} inserts errored: {errors[:3]}"
# Count persisted rows directly.
rows = app.db.query("SELECT COUNT(*) AS c FROM items")
# 100 concurrent inserts, all must persist (P2.2 regression:
# old py.detach+block_on path silently dropped ~49/50 writes).
assert rows[0]["c"] == 100, f"expected 100 rows, got {rows[0]['c']}"
finally:
if os.path.exists(db_path):
os.unlink(db_path)
os.rmdir(tmpdir)
@pytest.mark.asyncio
async def test_concurrent_writes_small_pool_no_saturation_collapse():
"""Regression: concurrent Python-handler writes must NOT collapse when the
pool is small (ADR-080).
The old request-scoped auto-transaction acquired a pool connection on the
async runtime and held it while the request queued on the single GIL
worker, then the handler's `app.db.query` acquired a *second* connection:
2N connections for N concurrent writes on an N-connection pool → immediate
saturation ("pool timed out", ~150 RPS at c=10 → ~3 RPS at c=11). With the
fix, concurrent writes stay flat and all persist.
"""
tmpdir = tempfile.mkdtemp()
db_path = os.path.join(tmpdir, "svc_small_pool.db")
open(db_path, "w").close()
try:
from justapi import Database
app = JustAPIApp()
# A deliberately tiny pool: 2 connections for 50 concurrent writers.
app.set_database(
Database(f"sqlite://{db_path}", max_connections=2, request_acquire_timeout=5.0)
)
@app.post("/items")
async def create(request):
body = request.json()
app.db.query(
"INSERT INTO items(name, qty) VALUES (?, ?)",
[body["name"], body["qty"]],
)
return {"ok": True}
async with AsyncTestClient(app, database=f"sqlite://{db_path}") as c:
app.db.query(
"CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT, qty INTEGER)"
)
import asyncio
async def insert_one(v):
return await c.post(
"/items", f'{{"name":"w{v}","qty":{v}}}'.encode()
)
results = await asyncio.gather(
*(insert_one(v) for v in range(50)),
return_exceptions=True,
)
# Every request must complete successfully (200) — the old code
# returned 500/503 "pool timed out" for the majority here.
errors = [
r for r in results if isinstance(r, Exception) or r.get("status") != 200
]
assert not errors, f"{len(errors)} writes failed: {errors[:3]}"
rows = app.db.query("SELECT COUNT(*) AS c FROM items")
assert rows[0]["c"] == 50, f"expected 50 rows, got {rows[0]['c']}"
finally:
if os.path.exists(db_path):
os.unlink(db_path)
os.rmdir(tmpdir)