Add owning-reversible TdsSyncClient over the blocking edge - #203
Conversation
9c1fc29 to
0ad00fe
Compare
There was a problem hiding this comment.
Pull request overview
Adds an owning, reversible synchronous row-fetch client over raw TCP while preserving async control-plane operations.
Changes:
- Adds
TdsSyncClientand async/blocking transport handoff. - Shares row-token handling across sync and async paths.
- Adds differential tests and mock mid-stream errors.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
mssql-tds/tests/test_sync_client.rs |
Tests sync parity, reversibility, and errors. |
mssql-tds/src/io/std_byte_source.rs |
Implements blocking TCP reads. |
mssql-tds/src/io/packet_buffer.rs |
Transfers buffered residual bytes. |
mssql-tds/src/io/blocking_reader.rs |
Supports seeded buffers and extraction. |
mssql-tds/src/io.rs |
Registers the blocking source. |
mssql-tds/src/datatypes/row_writer.rs |
Adds row-buffer recycling. |
mssql-tds/src/connection/transport/tds_transport.rs |
Defines blocking handoff hooks. |
mssql-tds/src/connection/transport/network_transport.rs |
Implements raw-TCP extraction/restoration. |
mssql-tds/src/connection/tds_sync_client.rs |
Implements synchronous fetching and reversal. |
mssql-tds/src/connection/tds_client.rs |
Adds conversion and shared token handling. |
mssql-tds/src/connection.rs |
Exports the sync client module. |
mssql-mock-tds/src/query_response.rs |
Models mid-stream server errors. |
mssql-mock-tds/src/protocol.rs |
Serializes injected error responses. |
Suppressed comments (1)
mssql-tds/src/connection/tds_sync_client.rs:156
self.activeis discarded when the wrapper is consumed, so reverting afternext_row_intoreturnedRowPausedorPlpPausedgives the async client anIdlecursor at a mid-row wire position. The PLP case is especially blocking because this sync API deliberately omitsread_active_plp_bytes, makinginto_asyncthe only way to continue. Move the sync pause state back into the inner client's active-row state before returning it.
let mut inner = self.inner;
inner
.transport
.restore_blocking_parts(tokio_stream, residual)?;
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| let std_stream = match boxed.into_blocking_std() { | ||
| Some(std_stream) => std_stream, | ||
| None => { | ||
| // Eligibility said raw TCP, so this is unreachable; if it ever | ||
| // fires the socket is already gone, so mark the connection dead | ||
| // rather than silently losing it. | ||
| self.known_dead = true; | ||
| return None; |
| pub fn into_sync(mut self) -> crate::connection::tds_sync_client::SyncConversion { | ||
| use crate::connection::tds_sync_client::{SyncConversion, TdsSyncClient}; | ||
|
|
||
| let runtime_handle = tokio::runtime::Handle::try_current().ok(); |
| inner, | ||
| reader, | ||
| empty_metadata: Vec::new(), | ||
| active: SyncRowState::Idle, |
| fn arm_deadline(&mut self) { | ||
| let deadline = self.request_timeout.map(|d| Instant::now() + d); | ||
| self.reader.source_mut().set_deadline(deadline); |
| let result = | ||
| drive_row_over_buffer_blocking(&mut self.reader, &context, resume.take(), writer)?; |
| match self.stream.read(buffer) { | ||
| Ok(n) => return Ok(n), |
| /// Creates a writer that reuses an existing (already-allocated) row buffer, | ||
| /// clearing it first. Lets batch fetchers recycle row allocations across a | ||
| /// result set instead of allocating one `Vec` per row. | ||
| pub fn from_recycled(mut row: Vec<ColumnValues>) -> Self { |
| token.put_u8(severity); | ||
|
|
||
| // Message (US_VARCHAR: u16 code-unit count + UTF-16LE) | ||
| token.put_u16_le(message.chars().count() as u16); |
Extend the L3 blocking sync decode to PLP string cells (varchar(max)/nvarchar(max)) alongside varbinary(max), and let a Login-Only-encryption connection (TLS dropped after login) expose its raw socket so it can flip to the synchronous fetch edge instead of being pinned to the async client.
0ad00fe to
56be7d1
Compare
|
Closing: the sans-I/O restructuring is not going to be productionized. The row-decode performance work that motivated much of this has been re-scoped around #247, where benchmarked spikes show the dominant win comes from a far smaller change — converting This is cleanup rather than a rejection of the analysis. The branch is deliberately not deleted, so the work remains recoverable if the direction is revisited. |
L4 — Owning-reversible
TdsSyncClientover the blocking edgeStacked on L3 (
dev/saurabh/sans-io-expose-l3-blocking-driver@7e11a11b) — base is L3, notmain.Surfaces the L3 blocking row driver as a public, reactor-free
TdsSyncClientso consumers (ODBCSQLFetch, later py-core) can drop their per-rowblock_onon the hot path. The asyncTdsClientsurface is unchanged.Additive-4 public surface (frozen)
TdsClient::into_sync(self) -> SyncConversion— opportunistic raw-TCP flip.SyncConversion { Converted(TdsSyncClient) | NotEligible(TdsClient) | Failed(TdsError) }. TLS transports return the async client intact viaNotEligible(not an error — caller keeps today'sblock_on).TdsSyncClient— owns the established transport + buffer over the blocking edge (concreteStdTcpByteSource; no genericBin the type name). 8 B-free fetch methods (next_row,next_row_into,fetch_rows_batch,take_info_messages,get_metadata,maybe_has_unread_rows,active_plp_reached_end,active_plp_collation).TdsSyncClient::into_async(self) -> TdsResult<TdsClient>— reverts the fd to the async client;Err(fd known-dead) on failed revert.Owning, not borrowed (regression tripwire): a
SyncRowFetcher<'a>borrowingdbc.clientcan't persist acrossSQLFetchFFI calls, so it would flip the fd per row on a 1M-row set. Owning storesTdsSyncClientby value → fd stays blocking across the whole result set (~2 flips/result-set, measured 13.569µs/pair).Dropdoes terminal clean-close only; the explicit async revert isinto_async— never a Drop-driven fd-revert.Invariants honored
tds_client.rsis the additiveinto_sync;cursor_ops.rsuntouched. The shared orchestration added alongside it ispub(crate)(see doc-note 3), so no public signature changes and no genericBleaks into any async signature.step_row/drive_row_over_buffer_blocking/plp_collect_step/ the L3 blocking driver are called as-is, not reimplemented.fetch_rows_batchauthored fresh as a thin loop over reactor-freenext_rowwith spare-vec recycling (no async batch twin, no Batch TDS row fetch to make mssql-odbc faster than msodbcsql18 #186 dependency).execute*/advance*/close*stay async (reached viainto_async). No cold-token blocking driver. AE PLP columns →UnimplementedFeature.read_active_plp_bytesdeferred to L5.Tests
Differential parity vs an all-async oracle (byte-identical), plus the named residual-straddle interleave gate in both directions (async→sync and sync→async where buffered
tds_read_bufferbytes straddle the flip mid-packet) and a clean-boundary negative control. The 7-cert lib baseline (4certificate_validator+ 3win_tls::validate, both fixture/cert-store-gated) is unchanged.Doc-notes (context, not a re-plan)
current_threadis fine). The multi_thread-deadlock concern applies only to the forbidden cold adapter; the earlier prototype multi_thread note was an echo-server co-host artifact (the real SQL peer is remote, so no local task is starved by a blocked thread).handle_row_read_tokenare factored into a singlepub(crate) TdsClient::apply_row_read_token(&mut self, token) -> TdsResult<TokenOutcome>(pluspub(crate) finalize_row_errorfor post-drain cleanup), living onTdsClientintds_client.rsso it has native access to private state (count_map,finalize_return_value/push_return_value, info buffer) with zero accessors and zero field-visibility churn. Both shells call it; only the ERROR drain is flavour-specific (TokenOutcome::DrainThenError→ asyncdrain_streamvs a blocking drain-to-DONE over L3, then the sharedfinalize_row_error).handle_row_read_token's async signature/visibility is byte-identical — only its body delegates. All connection/result-set state (metadata, INFO buffer, read-till-end flag) lives on the ownedinner; the wrapper keeps no mirrored copies, so the two clients cannot diverge. This keepsClientCore<B>unused (prefer-none) and the public-API diff empty while eliminating the parallel-handler drift risk.apply_row_read_token/finalize_row_errorowns (terminal DONE accounting + batch close, DONE-MORE accumulation, DONE error-flag → protocol error, ORDER no-op, INFO capture, RETURN_VALUE push, ERROR defer-drain-without-mutation, COLMETADATA usage error, andfinalize_row_errorbatch-clear + surfacedSqlServerError). The full pre-existing async suite passes with zero expected-output edits, so the shared-handler extraction is behavior-preserving.Info/EnvChange/SessionState/ReturnValue/ReturnStatus) through the sameapply_row_read_token, reaching a byte-identical terminal state to the asyncdrain_stream. An Error-token-mid-fetch case was added to both the differential set and the residual-straddle-interleave set (both directions), so the sync drain is exercised, not just the happy path.TdsSyncClientlives in a different module (tds_sync_client.rs) and callsinner.apply_row_read_token(...)cross-module,pub(crate)is the tightest reachable visibility (a privatefnis unreachable cross-module) — and only new additivepub(crate)methods were introduced. Zero pre-existing item was widened:capture_info_messagestays private (the earlier sync path that would have widened it now goes through the shared handler instead).