Skip to content

Commit 782ed7a

Browse files
RyanLamanskyRyanLamansky
committed
The TDS endpoint honors mid-stream attention: a carry-forward concurrent read doubles as the attention watcher, cancel/timeout abort the batch at statement/row safe points and acknowledge with DONE_ATTN only with transaction semantics matching real (survives cancel under XACT_ABORT OFF, rolls back under ON), WAITFOR interruptible via the connection-scoped execution cancellation that SimulatedDbCommand.Cancel() now also drives in-process, and the chunked OFFSET/FETCH backlog entry rewritten to record that real also redoes per-page work, demoting it to constant-factor perf polish.
1 parent ce255d4 commit 782ed7a

14 files changed

Lines changed: 587 additions & 45 deletions

File tree

‎CLAUDE.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ Per-feature deep-dives live under `docs/claude/`. Each entry below is a trigger:
165165
- **New top-level statement parser or dispatch-loop separator rules, double-quoted identifiers / QUOTED_IDENTIFIER** → [`grammar.md`](docs/claude/grammar.md) + [`control-flow.md`](docs/claude/control-flow.md).
166166
- **BACPAC import** (`Simulation.ImportBacpac` — multi-database via repeated calls, `BacpacImportOptions`, `ModelXmlReader` dispatcher, BCP wire format, `BacpacBuilder` test harness) → [`bacpac-loader.md`](docs/claude/bacpac-loader.md).
167167
- **Linked servers** (`Simulation.AddRemoteSimulation`, `sp_addlinkedserver` / `sp_dropserver`, four-part FROM routing through the remote's ADO.NET pipeline, `OPENQUERY(server,'query')` ad-hoc pass-through + compile-time schema discovery, `sys.servers`) → [`linked-servers.md`](docs/claude/linked-servers.md).
168-
- **TDS network endpoint** (`Simulation.ListenAsync` → `SimulatedNetworkListener`; real SqlClient over loopback TCP+TLS; SQLBatch/RPC/TM/BulkLoad (`SqlBulkCopy`); TVP (`0xF3` `SqlDbType.Structured`) RPC params via the shared `TdsColumnDecoder`; `sp_cursor*` API-server-cursor RPC family; credential enforcement via the `CREATE LOGIN` registry, Msg 18456 on mismatch; EF via plain `UseSqlServer`; oracles = `*.Tests.SqlClient` + `*.Tests.Smo`, the real-SMO consumer oracle) → [`tds-endpoint.md`](docs/claude/tds-endpoint.md).
168+
- **TDS network endpoint** (`Simulation.ListenAsync` → `SimulatedNetworkListener`; real SqlClient over loopback TCP+TLS; SQLBatch/RPC/TM/BulkLoad (`SqlBulkCopy`); TVP (`0xF3` `SqlDbType.Structured`) RPC params via the shared `TdsColumnDecoder`; `sp_cursor*` API-server-cursor RPC family; mid-stream attention (cancel / `CommandTimeout`) via a carry-forward concurrent read → `DONE_ATTN` ack + probed XACT_ABORT tx semantics; credential enforcement via the `CREATE LOGIN` registry, Msg 18456 on mismatch; EF via plain `UseSqlServer`; oracles = `*.Tests.SqlClient` + `*.Tests.Smo`, the real-SMO consumer oracle) → [`tds-endpoint.md`](docs/claude/tds-endpoint.md).
169169

170170
## Not modeled
171171

‎SqlServerSimulator.Tests.SqlClient/AttentionTests.cs‎

Lines changed: 257 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,257 @@
1+
using Microsoft.Data.SqlClient;
2+
using static Microsoft.VisualStudio.TestTools.UnitTesting.Assert;
3+
4+
namespace SqlServerSimulator;
5+
6+
/// <summary>
7+
/// Mid-stream attention (TDS cancel) handling: a client <c>SqlCommand.Cancel()</c>
8+
/// or an expiring <c>CommandTimeout</c> sends a TDS attention (packet type 6)
9+
/// while a batch is executing or streaming. The listener notices it at a safe
10+
/// point, aborts the batch, and replies with a DONE token carrying the
11+
/// DONE_ATTN flag; the session stays alive and reusable. SqlClient synthesizes
12+
/// the surfaced exception client-side (Msg -2 "Execution Timeout Expired" for a
13+
/// timeout, Msg 0 "Operation cancelled by user" for an explicit cancel) — the
14+
/// server emits no error token, only the acknowledgment. Semantics
15+
/// probe-confirmed against SQL Server 2025 (2026-07-18).
16+
/// </summary>
17+
[TestClass]
18+
public sealed class AttentionTests
19+
{
20+
public TestContext TestContext { get; set; } = null!;
21+
22+
private const string LongWait = "waitfor delay '00:00:30'";
23+
24+
/// <summary>Cancels the command shortly after execution starts, off the calling thread.</summary>
25+
private static void CancelAfter(SqlCommand command, int millis) =>
26+
_ = Task.Run(async () =>
27+
{
28+
await Task.Delay(millis);
29+
command.Cancel();
30+
});
31+
32+
private static async Task AssertSessionReusableAsync(SqlConnection connection, CancellationToken cancellationToken)
33+
{
34+
await using var probe = new SqlCommand("select 42", connection);
35+
AreEqual(42, await probe.ExecuteScalarAsync(cancellationToken));
36+
}
37+
38+
[TestMethod]
39+
public async Task CommandTimeout_OnWaitfor_RaisesTimeoutAndSessionStaysReusable()
40+
{
41+
var simulation = new Simulation();
42+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
43+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
44+
45+
await using (var command = new SqlCommand(LongWait, connection) { CommandTimeout = 1 })
46+
{
47+
var error = await ThrowsExactlyAsync<SqlException>(
48+
async () => await command.ExecuteNonQueryAsync(TestContext.CancellationToken));
49+
AreEqual(-2, error.Number);
50+
}
51+
52+
await AssertSessionReusableAsync(connection, TestContext.CancellationToken);
53+
}
54+
55+
[TestMethod]
56+
public async Task Cancel_DuringWaitfor_RaisesCancelAndSessionStaysReusable()
57+
{
58+
var simulation = new Simulation();
59+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
60+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
61+
62+
await using (var command = new SqlCommand(LongWait, connection) { CommandTimeout = 0 })
63+
{
64+
CancelAfter(command, 200);
65+
var error = await ThrowsExactlyAsync<SqlException>(
66+
async () => await command.ExecuteNonQueryAsync(TestContext.CancellationToken));
67+
AreEqual(0, error.Number);
68+
}
69+
70+
await AssertSessionReusableAsync(connection, TestContext.CancellationToken);
71+
}
72+
73+
[TestMethod]
74+
public async Task Cancel_MidLargeResult_AbortsAndSessionStaysReusable()
75+
{
76+
var simulation = new Simulation();
77+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
78+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
79+
80+
const string bigResult = "select value, replicate('x', 200) from generate_series(1, 500000)";
81+
await using (var command = new SqlCommand(bigResult, connection) { CommandTimeout = 0 })
82+
{
83+
CancelAfter(command, 50);
84+
_ = await ThrowsExactlyAsync<SqlException>(async () =>
85+
{
86+
await using var reader = await command.ExecuteReaderAsync(TestContext.CancellationToken);
87+
while (await reader.ReadAsync(TestContext.CancellationToken))
88+
{
89+
}
90+
});
91+
}
92+
93+
await AssertSessionReusableAsync(connection, TestContext.CancellationToken);
94+
}
95+
96+
[TestMethod]
97+
public async Task Cancel_WithOpenTransaction_DefaultXactAbort_LeavesTransactionOpen()
98+
{
99+
var simulation = new Simulation();
100+
Wire.ExecInProc(simulation, "create table t (id int)");
101+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
102+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
103+
104+
await using (var begin = new SqlCommand("begin tran; insert t values (1)", connection))
105+
_ = await begin.ExecuteNonQueryAsync(TestContext.CancellationToken);
106+
107+
await using (var command = new SqlCommand(LongWait, connection) { CommandTimeout = 0 })
108+
{
109+
CancelAfter(command, 200);
110+
_ = await ThrowsExactlyAsync<SqlException>(
111+
async () => await command.ExecuteNonQueryAsync(TestContext.CancellationToken));
112+
}
113+
114+
// XACT_ABORT OFF (the default): the transaction survives the cancel.
115+
await using (var trancount = new SqlCommand("select @@trancount", connection))
116+
AreEqual(1, await trancount.ExecuteScalarAsync(TestContext.CancellationToken));
117+
await using (var rows = new SqlCommand("select count(*) from t", connection))
118+
AreEqual(1, await rows.ExecuteScalarAsync(TestContext.CancellationToken));
119+
120+
await using (var rollback = new SqlCommand("rollback", connection))
121+
_ = await rollback.ExecuteNonQueryAsync(TestContext.CancellationToken);
122+
await using var afterRollback = new SqlCommand("select count(*) from t", connection);
123+
AreEqual(0, await afterRollback.ExecuteScalarAsync(TestContext.CancellationToken));
124+
}
125+
126+
[TestMethod]
127+
public async Task Cancel_WithOpenTransaction_XactAbortOn_RollsTransactionBack()
128+
{
129+
var simulation = new Simulation();
130+
Wire.ExecInProc(simulation, "create table t (id int)");
131+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
132+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
133+
134+
await using (var begin = new SqlCommand("set xact_abort on; begin tran; insert t values (1)", connection))
135+
_ = await begin.ExecuteNonQueryAsync(TestContext.CancellationToken);
136+
137+
await using (var command = new SqlCommand(LongWait, connection) { CommandTimeout = 0 })
138+
{
139+
CancelAfter(command, 200);
140+
_ = await ThrowsExactlyAsync<SqlException>(
141+
async () => await command.ExecuteNonQueryAsync(TestContext.CancellationToken));
142+
}
143+
144+
// XACT_ABORT ON: the cancel rolls the transaction back.
145+
await using (var trancount = new SqlCommand("select @@trancount", connection))
146+
AreEqual(0, await trancount.ExecuteScalarAsync(TestContext.CancellationToken));
147+
await using var rows = new SqlCommand("select count(*) from t", connection);
148+
AreEqual(0, await rows.ExecuteScalarAsync(TestContext.CancellationToken));
149+
}
150+
151+
[TestMethod]
152+
public async Task Cancel_MidBatch_DiscardsRemainingStatements()
153+
{
154+
var simulation = new Simulation();
155+
Wire.ExecInProc(simulation, "create table t (id int)");
156+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
157+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
158+
159+
// The first statement commits (autocommit); the cancel fires during the
160+
// WAITFOR, so the batch aborts at the statement boundary and the third
161+
// statement never runs — statement-boundary atomicity.
162+
var batchText = $"insert t values (1); {LongWait}; insert t values (2)";
163+
await using (var batch = new SqlCommand(batchText, connection) { CommandTimeout = 0 })
164+
{
165+
CancelAfter(batch, 200);
166+
_ = await ThrowsExactlyAsync<SqlException>(
167+
async () => await batch.ExecuteNonQueryAsync(TestContext.CancellationToken));
168+
}
169+
170+
await using var rows = new SqlCommand("select count(*) from t", connection);
171+
AreEqual(1, await rows.ExecuteScalarAsync(TestContext.CancellationToken));
172+
}
173+
174+
[TestMethod]
175+
public async Task Cancel_DuringParameterizedRpc_AbortsAndSessionStaysReusable()
176+
{
177+
var simulation = new Simulation();
178+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
179+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
180+
181+
await using (var command = new SqlCommand($"select @p; {LongWait}", connection) { CommandTimeout = 0 })
182+
{
183+
_ = command.Parameters.AddWithValue("@p", 7);
184+
CancelAfter(command, 200);
185+
_ = await ThrowsExactlyAsync<SqlException>(async () =>
186+
{
187+
await using var reader = await command.ExecuteReaderAsync(TestContext.CancellationToken);
188+
do
189+
{
190+
while (await reader.ReadAsync(TestContext.CancellationToken))
191+
{
192+
}
193+
}
194+
while (await reader.NextResultAsync(TestContext.CancellationToken));
195+
});
196+
}
197+
198+
await AssertSessionReusableAsync(connection, TestContext.CancellationToken);
199+
}
200+
201+
[TestMethod]
202+
public async Task Cancel_RacingNaturalCompletion_DoesNotHang()
203+
{
204+
var simulation = new Simulation();
205+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
206+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
207+
208+
// Fire a cancel around a query that completes almost immediately: the
209+
// attention may land before, during, or after natural completion. Every
210+
// outcome must acknowledge without hanging, and the session must survive.
211+
for (var i = 0; i < 30; i++)
212+
{
213+
await using var command = new SqlCommand("select 1", connection) { CommandTimeout = 5 };
214+
CancelAfter(command, 0);
215+
try
216+
{
217+
await using var reader = await command.ExecuteReaderAsync(TestContext.CancellationToken);
218+
while (await reader.ReadAsync(TestContext.CancellationToken))
219+
{
220+
}
221+
}
222+
catch (SqlException)
223+
{
224+
// A cancel that landed mid-flight surfaces here; tolerated.
225+
}
226+
catch (InvalidOperationException)
227+
{
228+
// A cancel that beat ExecuteReaderAsync to the starting line is
229+
// rejected client-side by SqlClient before anything reaches the
230+
// server; also a valid race outcome.
231+
}
232+
}
233+
234+
await AssertSessionReusableAsync(connection, TestContext.CancellationToken);
235+
}
236+
237+
[TestMethod]
238+
public async Task SequentialCommands_AfterCancel_AllSucceed()
239+
{
240+
var simulation = new Simulation();
241+
await using var listener = await simulation.ListenAsync(0, TestContext.CancellationToken);
242+
await using var connection = await Wire.OpenAsync(listener, TestContext.CancellationToken);
243+
244+
await using (var command = new SqlCommand(LongWait, connection) { CommandTimeout = 0 })
245+
{
246+
CancelAfter(command, 200);
247+
_ = await ThrowsExactlyAsync<SqlException>(
248+
async () => await command.ExecuteNonQueryAsync(TestContext.CancellationToken));
249+
}
250+
251+
for (var i = 1; i <= 5; i++)
252+
{
253+
await using var command = new SqlCommand($"select {i}", connection);
254+
AreEqual(i, await command.ExecuteScalarAsync(TestContext.CancellationToken));
255+
}
256+
}
257+
}

‎SqlServerSimulator.Tests/WaitForDelayTests.cs‎

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,10 @@
55
namespace SqlServerSimulator;
66

77
/// <summary>
8-
/// Tests for <c>WAITFOR DELAY</c>. The simulator uses
9-
/// <see cref="Thread.Sleep(TimeSpan)"/> on the calling thread, matching real
10-
/// SQL Server's "blocks the connection" semantics. To keep CI fast, only
11-
/// one test actually sleeps a non-trivial duration; everything else
8+
/// Tests for <c>WAITFOR DELAY</c>. The simulator blocks the calling thread on
9+
/// a cancellable wait, matching real SQL Server's "blocks the connection"
10+
/// semantics while staying interruptible by a command cancel. To keep CI fast,
11+
/// only one test actually sleeps a non-trivial duration; everything else
1212
/// exercises parsing / dispatch / skip-mode / error paths with a
1313
/// <c>'00:00:00'</c> operand. Behavior probed against SQL Server 2025
1414
/// (2026-05-11).
@@ -60,6 +60,40 @@ public void Delay_50ms_ActuallySleeps()
6060
$"Expected well under a misparsed-magnitude sleep, got {elapsed.TotalMilliseconds}ms");
6161
}
6262

63+
/// <summary>
64+
/// In-process <c>Cancel()</c> from another thread interrupts a running
65+
/// <c>WAITFOR DELAY</c> — the same abort machinery the TDS attention path
66+
/// drives, exposed through the ADO surface. The 30-second wait returns
67+
/// promptly once cancelled; the connection stays usable.
68+
/// </summary>
69+
[TestMethod]
70+
public void Delay_InterruptedByInProcessCancel_ReturnsPromptly()
71+
{
72+
var sim = new Simulation();
73+
using var connection = sim.CreateDbConnection();
74+
connection.Open();
75+
using var command = connection.CreateCommand();
76+
command.CommandText = "waitfor delay '00:00:30'";
77+
var canceller = new Thread(() =>
78+
{
79+
Thread.Sleep(200);
80+
command.Cancel();
81+
});
82+
83+
var start = Stopwatch.GetTimestamp();
84+
canceller.Start();
85+
_ = command.ExecuteNonQuery();
86+
canceller.Join();
87+
var elapsed = Stopwatch.GetElapsedTime(start);
88+
89+
IsLessThan(10000, elapsed.TotalMilliseconds,
90+
$"Expected the cancel to interrupt the 30s wait promptly, got {elapsed.TotalMilliseconds}ms");
91+
92+
using var probe = connection.CreateCommand();
93+
probe.CommandText = "select 42";
94+
AreEqual(42, probe.ExecuteScalar());
95+
}
96+
6397
[TestMethod]
6498
public void Delay_InUntakenIf_DoesNotSleep()
6599
{

‎SqlServerSimulator/Network/TdsSession.Rpc.cs‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,10 @@ private async ValueTask ExecuteStatementRpcAsync(
182182
outputs.Add((i, boundParameters[i], bound));
183183
}
184184

185-
await this.StreamOutcomesAsync(command, writer, Tds.TokenDoneInProc, trailingTokensFollow: true, cancellationToken).ConfigureAwait(false);
185+
// A cancelled RPC skips its trailing RETURNSTATUS / RETURNVALUE /
186+
// DONEPROC: the batch loop emits the single DONE_ATTN acknowledgment.
187+
if (await this.StreamOutcomesAsync(command, writer, Tds.TokenDoneInProc, trailingTokensFollow: true, cancellationToken).ConfigureAwait(false))
188+
return;
186189

187190
writer.WriteReturnStatus(0);
188191
if (handleReturn is { } handleValue)
@@ -219,7 +222,10 @@ private async ValueTask ExecuteProcedureRpcAsync(TdsRpcRequest request, TdsToken
219222
outputs.Add((i, request.Parameters[i], bound));
220223
}
221224

222-
await this.StreamOutcomesAsync(command, writer, Tds.TokenDoneInProc, trailingTokensFollow: true, cancellationToken).ConfigureAwait(false);
225+
// A cancelled RPC skips its trailing RETURNSTATUS / RETURNVALUE /
226+
// DONEPROC: the batch loop emits the single DONE_ATTN acknowledgment.
227+
if (await this.StreamOutcomesAsync(command, writer, Tds.TokenDoneInProc, trailingTokensFollow: true, cancellationToken).ConfigureAwait(false))
228+
return;
223229

224230
writer.WriteReturnStatus(returnParameter.Value is int returnCode ? returnCode : 0);
225231
foreach (var (ordinal, wire, bound) in outputs)

0 commit comments

Comments
 (0)