Repository navigation
Faster, non-blocking /state responses - #383
Conversation
Records that share a type are dumped with a cached TypeAdapter(list[T]) instead of one dump_json per record, and the bytes go straight into the Response without a decode. Mixed subclasses keep the per-record path. Output is byte-identical. Signed-off-by: Emily <[email protected]>
288fafc to
86693e8
Compare
Test Results1 178 tests +63 1 168 ✅ +63 10m 8s ⏱️ +19s Results for commit ca788b3. ± Comparison against base commit 652b6a6. This pull request removes 1 and adds 64 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #383 +/- ##
==========================================
+ Coverage 88.37% 88.60% +0.23%
==========================================
Files 165 167 +2
Lines 20875 21465 +590
Branches 1914 1948 +34
==========================================
+ Hits 18448 19020 +572
- Misses 2427 2445 +18 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Records of one GatewayStruct type go to pydantic-core in calls of up to 5k records, rather than one call per record that calls back into Python through _drop_implicitly_unset and _serialize_timestamp. Unset fields become a native exclude, and the timestamp callback is skipped for runs of plain naive datetimes. Output stays byte-identical to dumping each record; structs where native exclusion could differ fall back to that. Also replaces the list[T] TypeAdapter from the previous commit, which dropped the struct's own config (ser_json_inf_nan, and temporals inside Any or dict fields). Signed-off-by: Emily <[email protected]>
The /state handlers are now sync, so FastAPI runs the query and its serialization in its threadpool instead of holding the event loop: one large state no longer stalls every other request until it finishes. DuckDBState.query now holds _db_lock around the SELECT and the object store lookup. An engine-thread insert crossing the buffer threshold runs load_data on the same cursor; interleaved, DuckDB raises, _StateManager returns [] for the query, and the buffered insert batch is lost. Signed-off-by: Emily <[email protected]>
emilybarrettp72
changed the title
1. Preserve Model-Owned AdaptersMajor. Comment location The fast path calls from pydantic import TypeAdapter
from csp_gateway import GatewayStruct
from csp_gateway.server.web.routes.shared import prepare_response
from csp_gateway.utils.struct.base import type_adapter_for
class RedactingAdapter(TypeAdapter):
def dump_json(self, instance, **kwargs):
return super().dump_json(instance, exclude={"secret"}, **kwargs)
class Record(GatewayStruct):
public: str = "visible"
secret: str = "private"
@classmethod
def type_adapter(cls):
return RedactingAdapter(cls)
record = Record.from_field_values()
expected = b"[" + type_adapter_for(record).dump_json(record) + b"]"
actual = prepare_response([record], is_list_model=True).body
print("Per record:", expected.decode())
print("Batched: ", actual.decode())
assert actual == expected, "model-owned adapter was bypassed"Observed output before the assertion: 2. Preserve Nested Alias ConfigurationMajor. Comment location Passing the parent's from pydantic import ConfigDict, Field
from csp_gateway import GatewayStruct
from csp_gateway.server.web.routes.shared import prepare_response
from csp_gateway.utils.struct.base import type_adapter_for
class Child(GatewayStruct):
model_config = ConfigDict(serialize_by_alias=True)
value: int = Field(default=1, serialization_alias="renamed")
class Parent(GatewayStruct):
child: Child
record = Parent.from_field_values(child=Child.from_field_values())
expected = b"[" + type_adapter_for(record).dump_json(record) + b"]"
actual = prepare_response([record], is_list_model=True).body
print("Per record:", expected.decode())
print("Batched: ", actual.decode())
assert actual == expected, "nested alias configuration was overridden"Observed output before the assertion: 3. Include Computed Fields in the Collision GuardMinor. Comment location The duplicate-output-key check only considers regular fields. A computed field from pydantic import ConfigDict, computed_field
from csp_gateway import GatewayStruct
from csp_gateway.server.web.routes.shared import prepare_response
from csp_gateway.utils.struct.base import type_adapter_for
class Record(GatewayStruct):
model_config = ConfigDict(serialize_by_alias=True)
value: int
@computed_field(alias="value")
@property
def computed(self) -> int:
return 42
record = Record.from_field_values()
expected = b"[" + type_adapter_for(record).dump_json(record) + b"]"
actual = prepare_response([record], is_list_model=True).body
print("Per record:", expected.decode())
print("Batched: ", actual.decode())
assert actual == expected, "computed-field alias escaped unset-field removal"Observed output before the assertion: 4. Require Overlap in the Concurrency TestMinor, test coverage. Comment location This test can pass without querying while the writer is active. If the writer import threading
from importlib import import_module
from unittest.mock import patch
tests = import_module("csp_gateway.tests.server.gateway.csp.test_state")
state_module = import_module("csp_gateway.server.gateway.csp.state")
original_start = threading.Thread.start
original_query = state_module.DuckDBState.query
writer_finished = threading.Event()
query_overlaps = []
def start_writer_first(thread):
original_start(thread)
thread.join(timeout=10)
assert not thread.is_alive(), "writer did not finish within the bound"
writer_finished.set()
def observe_query(state, *args, **kwargs):
query_overlaps.append(not writer_finished.is_set())
return original_query(state, *args, **kwargs)
with (
patch.object(threading.Thread, "start", start_writer_first),
patch.object(state_module.DuckDBState, "query", observe_query),
):
tests.test_duckdb_query_concurrent_with_buffered_inserts()
assert query_overlaps == [False]
print("Existing concurrency test: passed")
print("Queries during writer: ", sum(query_overlaps))
print("Final count queries: ", len(query_overlaps))Observed output: |
- Structs with their own type_adapter take the per-record path, so an adapter that changes the output (e.g. redacts a field) is honoured. - Pass the same by_alias TypeAdapter.dump_json passes (None on current pydantic), so nested models keep their own serialize_by_alias rather than inheriting the parent's. - Count computed-field output keys in the shared-key guard: popping an unset field's key also dropped a computed field emitted under it. - The DuckDB concurrency test now parks the writer mid-load until a query starts and asserts the overlap happened. - Add AsyncClient tests against a live gateway: every return type and concurrent state calls match the per-record output. Signed-off-by: Emily <[email protected]>
timkpaine
left a comment
timkpaine
left a comment
There was a problem hiding this comment.
1. Handle Plain Field Validators in the Batch Guard
Major. Comment location
_serializers_reached_by_none() assumes every function-* schema contains an
inner schema. A plain field validator produces a function-plain node without
that key. For a field declared without an explicit default, the batch guard
raises KeyError('schema') even when the record supplies a valid value.
Per-record serialization succeeds. The same payload changes /state from HTTP
200 to 500, and the websocket serialization path fails too. Please treat plain
validators as terminal nodes or fall back to per-record serialization.
Standalone repro, run from the PR checkout at 88746524:
from pydantic import field_validator
from csp_gateway import GatewayStruct
from csp_gateway.server.web.routes.shared import prepare_response
from csp_gateway.utils.struct.base import type_adapter_for
class Record(GatewayStruct):
value: int
@field_validator("value", mode="plain")
@classmethod
def parse_value(cls, value):
return int(value)
record = Record.from_field_values(value="3")
expected = b"[" + type_adapter_for(record).dump_json(record) + b"]"
print("Per record:", expected.decode())
actual = prepare_response([record], is_list_model=True).body
assert actual == expectedObserved result, with the traceback shortened:
Per record: [{"value":3}]
KeyError: 'schema'
2. Synchronize at the Query/Insert Boundary
Minor, test reliability. Comment location
The revised test prevents the old zero-query false pass, but query_started
is signaled before entering the query and overlap is counted only after it
returns. If the writer finishes between those points, the test fails despite
correct production locking. Please coordinate at the actual SELECT/insert
boundary rather than relying on the writer still being alive afterward.
This repro runs the existing test and forces that valid scheduling order: after
the query-start signal, let the writer finish before calling the original query.
It leaves the production locks intact and uses a bounded join, not a sleep.
Standalone repro, run from the PR checkout at 88746524 with test dependencies:
import threading
from importlib import import_module
from unittest.mock import patch
tests = import_module("csp_gateway.tests.server.gateway.csp.test_state")
state_module = import_module("csp_gateway.server.gateway.csp.state")
original_start = threading.Thread.start
original_query = state_module.DuckDBState.query
writers = []
def capture_writer(thread):
writers.append(thread)
return original_start(thread)
def writer_finishes_first(state, *args, **kwargs):
writer = writers[0]
writer.join(timeout=10)
assert not writer.is_alive(), "writer did not finish within the bound"
rows = original_query(state, *args, **kwargs)
print("Rows returned:", len(rows))
return rows
with (
patch.object(threading.Thread, "start", capture_writer),
patch.object(state_module.DuckDBState, "query", writer_finishes_first),
):
tests.test_duckdb_query_concurrent_with_buffered_inserts()Observed result, omitting an unrelated schema warning and shortening the traceback:
Rows returned: 5000
AssertionError: no query ran while the writer was inserting
Both repros were verified with Python 3.11.17 and Pydantic 2.14.0. The exceptions
above are intentional: the first demonstrates the response crash; the second
demonstrates a false test failure, not a production locking defect.
Signed-off-by: Tim Paine <[email protected]>
2 participants
/stateoff the event loop.main.Changes (one commit each)
prepare_responsesends bytes straight toResponse(no.decode()).GatewayStruct._dump_json_list):exclude.datetime.bytes, so the final join releases the GIL.Noneextra="allow"_implicitly_unsetTypeAdapter(list[T])path, which lost the struct's config forser_json_inf_nanand for temporals insideAny/dictfields (e.g./controls/statsnow)./statehandlers are sync, so FastAPI runs them in its threadpool.DuckDBState.querynow holds_db_lockaround the SELECT and the object-store lookup.main: an engine-thread insert crossing the buffer threshold shares the query's cursor. The DuckDB error is swallowed into[], and the insert batch is lost.Benchmarks
MountRestRoutes, DuckDB state), Trading-like structs, seed 42.mainand this PR alternating on the same box.GET /api/v1/state/<channel>over localhost HTTP, median of 5.mainin every cell (every rep, client and round).Concurrent clients, 1M records, 4 simultaneous full queries, median of 5 rounds:
csp engine lateness (10 ms timer, worst of 3 rounds, 1M records, max / p99):
main(Struct1 max 150 → 255 ms; Struct2 p99 70 → 110 ms). Onmainthose queries run one at a time on the event loop; here they run in parallel threads that trade the GIL with the engine. Not addressed in this PR.Verification
main's per-record output cover:NonefieldsNaT/subclassed timestampsAny/dictfieldsextra, mixed subclasses, chunk boundariesGatewayStructsubclasses ininternal_librarytake the fast path and matchmain(350 cases).main: 369/staterequests (statuses, headers, bodies, error paths, OpenAPI) are identical.main./statenot blocking the event loopAsyncClientagainst a live gateway (every return type, filtered and concurrent/statecalls) matchesmain's per-record output.type_adapter, nestedserialize_by_alias, computed-field output keys; each has a test that fails without the fix.main, only for data that fails or warns: