Explore timing over the fake driver; a paused query holds back what a call in flight brings back

.github/workflows/test.yml

@@ -377,6 +377,12 @@ jobs:
         }
         "
 
+    # no SQL Server here, but the event contract can still be checked: each recorded native
+    # session (test/contract/native) is replayed through the real JS layer over a fake driver
+    - name: Event contract, replayed over the fake driver (no SQL Server needed)
+      working-directory: node_modules/msnodesqlv8
+      run: npx mocha --exit test/contract/replay.test.js
+
     - name: Verify macOS build success
       working-directory: node_modules/msnodesqlv8
       run: |

lib/reader.js

@@ -340,6 +340,8 @@ class Query extends BasePromises {
 
   moveToNextResult (nextResultSetInfo) {
     setImmediate(() => {
+      // a pause that landed while the call was in flight: nothing more until the resume
+      if (this.park(() => this.moveToNextResult(nextResultSetInfo))) return
       if (this.cancelled) {
         this.running = false
         logger.debugLazy(() => `[${this.notify.getHandle()}] query ${this.queryId} has been cancelled - ending query ${this.queryId} `, this.context)
@@ -403,6 +405,12 @@ class Query extends BasePromises {
     return true
   }
 
+  // an error from a call that was in flight when the pause landed: the error, done and
+  // free all wait for the resume, and the paused query keeps holding the connection
+  endOrPark (err) {
+    if (!this.park(() => this.end(err))) this.end(err)
+  }
+
   // the native statement of this paused query has been cancelled. what is parked will
   // not run, and nothing more is emitted but the end: a step still in flight ends the
   // query when it returns.
@@ -434,7 +442,7 @@ class Query extends BasePromises {
       this.continueBatch()
     }).catch(err => {
       logger.debugLazy(() => `dispatch error for queryId ${this.queryId}: ${err}`, 'DriverRead.dispatch', this.context)
-      this.end(err)
+      this.endOrPark(err)
     })
   }
 
@@ -462,37 +470,44 @@ class Query extends BasePromises {
       .then(nextResultSetInfo => {
         this.moveToNextResult(nextResultSetInfo)
       }).catch(err => {
-        this.end(err)
+        this.endOrPark(err)
       })
   }
 
+  // the query has executed: report what came with it, then read on
+  begun (res) {
+    if (res.warning) {
+      this.routeStatementError(res.warning, this.callback, this.notify, !res.queryResult.endOfResults)
+      if (!this.cancelled) {
+        res.warning.forEach(err => {
+          if (!this.cancelled) {
+            this.cancelled = err.message.includes('Operation canceled')
+            if (this.cancelled) {
+              logger.debugLazy(() => `statement has been cancelled queryId ${this.queryId}`, this.context)
+            }
+          }
+        })
+      }
+    }
+    this.outputParams = res.outputParams
+    this.meta = res.queryResult.meta
+    this.rowCount = res.queryResult.rowCount || 0
+    this.currentQueryResult = res.queryResult
+    if (this.meta.length > 0) {
+      this.notify.emit('meta', this.meta)
+      this.dispatch()
+    } else {
+      this.nextResult()
+    }
+  }
+
   begin () {
     this.beginQuery(this.queryId, this.query, this.params).then(res => {
+      // the handle at once, so a cancel can reach the statement; what is emitted waits
+      // for a resume, if the query was paused while it executed
       this.notify.setHandle(res.queryResult.handle)
-
-      if (res.warning) {
-        this.routeStatementError(res.warning, this.callback, this.notify, !res.queryResult.endOfResults)
-        if (!this.cancelled) {
-          res.warning.forEach(err => {
-            if (!this.cancelled) {
-              this.cancelled = err.message.includes('Operation canceled')
-              if (this.cancelled) {
-                logger.debugLazy(() => `statement has been cancelled queryId ${this.queryId}`, this.context)
-              }
-            }
-          })
-        }
-      }
-      this.outputParams = res.outputParams
-      this.meta = res.queryResult.meta
-      this.rowCount = res.queryResult.rowCount || 0
-      this.currentQueryResult = res.queryResult
-      if (this.meta.length > 0) {
-        this.notify.emit('meta', this.meta)
-        this.dispatch()
-      } else {
-        this.nextResult()
-      }
+      const begun = () => this.begun(res)
+      if (!this.park(begun)) begun()
     }).catch(err => {
       logger.debugLazy(() => `dispatch error for queryId ${this.queryId}: ${err}`, this.context)
       // Don't call end() here - it's already handled in beginQuery for binding errors

test/common/fake-native.js

@@ -7,7 +7,9 @@
 //
 // results are released in causal order: a recorded result is delivered only once JS has
 // made every call recorded before it, one result per tick. so a query whose recorded result
-// is HY008 waits until JS has actually sent the cancel - no timestamps needed.
+// is HY008 waits until JS has actually sent the cancel. and a result is held for as long
+// as the driver took to give it, counted from when JS made the call, so a scenario that
+// acts on a timer (pause after 100 ms) sees the timing the server gave it.
 //
 // it also checks, whatever the log says, that JS respects the driver:
 //   overlap        a second call (cancel apart) while one is still in flight on the connection
@@ -22,10 +24,17 @@ const { decode } = require('../contract/native-tap')
 
 const STALL_MS = 1000
 
+// delay: optional (entry, recordedMs) => ms, how long to hold each recorded state or result,
+// counted from when JS made its call (default: recordedMs, as long as the driver took). the
+// explorer uses it to vary timing against the scenario's own timers - what comes back stays
+// what the driver sent.
 class ReplayNative {
-  constructor (log, { stallMs = STALL_MS } = {}) {
+  constructor (log, { stallMs = STALL_MS, delay = null } = {}) {
     this.log = log
     this.stallMs = stallMs
+    this.delay = delay
+    this.calledAt = new Map() // recorded call id -> when JS made it
+    this.callById = new Map(log.filter(e => e.ev === 'call').map(e => [e.id, e]))
     this.consumed = new Set() // recorded call ids JS has made
     this.callbacks = new Map() // recorded call id -> JS callback
     this.stateCallbacks = new Map() // recorded call id -> JS stateChangeCallback
@@ -78,6 +87,7 @@ class ReplayNative {
     const entry = this.match(name, callArgs)
     if (!entry) return
     this.consumed.add(entry.id)
+    this.calledAt.set(entry.id, performance.now())
     if (cb) this.callbacks.set(entry.id, { name, cb })
     if (name !== 'cancelQuery' && cb) this.inFlight.set(entry.id, name)
     const queryObj = callArgs[1]
@@ -102,16 +112,29 @@ class ReplayNative {
         continue
       }
       this.delivering = true
-      setImmediate(() => {
+      const next = () => {
         this.delivering = false
         this.cursor++
         this.deliver(e)
         this.pump()
-      })
+      }
+      const ms = this.holdFor(e)
+      if (ms > 0) setTimeout(next, ms)
+      else setImmediate(next)
       return
     }
   }
 
+  // how much longer to hold a recorded state or result: as long as the driver took to give
+  // it after the call (or what the delay option makes of that), less the time already passed
+  holdFor (e) {
+    const call = this.callById.get(e.of)
+    const recorded = call && typeof e.t === 'number' && typeof call.t === 'number' ? Math.max(0, e.t - call.t) : 0
+    const target = this.delay ? this.delay(e, recorded) : recorded
+    const since = performance.now() - (this.calledAt.get(e.of) ?? performance.now())
+    return Math.max(0, Math.round(target - since))
+  }
+
   deliver (e) {
     if (e.ev === 'state') {
       const onState = this.stateCallbacks.get(e.of)
@@ -129,7 +152,7 @@ class ReplayNative {
     this.stallTimer = setTimeout(() => {
       const waiting = [...this.inFlight.values()]
       const blocked = this.log[this.cursor]
-      if (waiting.length > 0 && blocked) {
+      if (waiting.length > 0 && blocked?.ev === 'call' && !this.consumed.has(blocked.id)) {
         this.violation('stalled', `JS waits on ${waiting.join(', ')}; the log waits on ${blocked.name ?? blocked.ev}(${JSON.stringify(blocked.args?.[0] ?? '')})`)
       }
     }, this.stallMs)

test/contract/HANDOVER.md

@@ -102,8 +102,13 @@ tests `require('msnodesqlv8')`), `npx node-gyp rebuild`, copy `.env-cmdrc`, syml
    (`queue/paused-then-cancel-queued-before`), and submit while paused (`queue/paused-then-submit`).
 3. `done` on every path (#479, the errors), then #478.
 
-Open decision: a pause that lands while a native call is in flight still lets that call's
-events out, e.g. the next result set's `meta`. Should paused mean no events at all?
+Settled (2026-10-17): a paused query emits nothing until resumed, including what a native
+call that was in flight when the pause landed brings back: the `meta` after execute or after
+`nextResultSet`, rows, and an error from a fetch or `nextResultSet` (with its `done` and
+`free`). The explorer found the old behaviour: `pause/before-execute` gave `meta` before or
+after the resume, depending on timing. Corpus: `pause/during-execute-then-{result,error}`,
+`pause/during-fetch`. Still not held: an error *at execute* while paused, because `beginQuery`
+queues the free before it rejects, so holding the error would let `free` out first.
 
 ## State at first handover
 

test/contract/README.md

@@ -62,6 +62,24 @@ npx mocha test/contract/replay.test.js           # all of them, ~20 s
 When a change to the JS is meant to change the native calls (not just the events), record
 again. The recordings are per platform, because the drivers differ (#483).
 
+Each result is held for as long as the driver took to give it, so a scenario that acts on a
+timer (pause after 100 ms) replays with the timing the server gave it.
+
+## Exploring timing
+
+`explore.js` replays each recording many times, holding each result for its recorded
+latency, give or take a fifth, plus a few ms. The scenario's own timers and the JS layer's
+steps then land at different points against the driver's results. A contract has to hold
+whatever the timing: every run must match the corpus with no violations. A `timed`
+scenario, one whose pause comes from a timer, is compared by shape and total rows, since
+how the rows split around the pause is timing.
+
+```sh
+node test/contract/explore.js --runs 20 pause/        # 20 timings per scenario
+node test/contract/explore.js --scale 10 cancel/      # wider jitter
+node test/contract/explore.js --seed 7 --runs 1 x     # replay exactly the timing of a failure
+```
+
 ## Status of a transcript
 
 Each file has `# status: unreviewed` or `# status: agreed`.

test/contract/capture.js

@@ -49,7 +49,7 @@ async function main () {
     const variants = await captureScenario(opts.root, connectionString, scenario, opts.repeat)
     const expected = corpus.read(scenario.name)
     if (variants.length > 1) summary.nondeterministic.push(`${scenario.name} (${variants.map(v => v.count).join('/')})`)
-    const allMatch = expected && variants.every(v => corpus.matches(expected, v.lines))
+    const allMatch = expected && variants.every(v => corpus.matchesScenario(scenario, expected, v.lines))
     const mark = !expected ? 'NEW ' : allMatch ? 'same' : 'DIFF'
     if (!expected) summary.missing++
     else if (allMatch) summary.same++
@@ -60,7 +60,7 @@ async function main () {
       for (const v of variants) {
         const indent = s => s.split('\n').map(l => `    ${l}`).join('\n')
         console.log(`  -- seen ${v.count}x`)
-        if (expected && !corpus.matches(expected, v.lines)) console.log(indent(corpus.diff(expected.variants[0], v.lines)))
+        if (expected && !corpus.matchesScenario(scenario, expected, v.lines)) console.log(indent(corpus.diff(expected.variants[0], v.lines)))
         else console.log(indent(v.lines.join('\n')))
       }
     }
@@ -85,7 +85,7 @@ async function recordNative (root, connectionString, scenario) {
   fs.mkdirSync(path.dirname(file), { recursive: true })
   const lines = await runIsolated(root, connectionString, scenario, tmp)
   const expected = corpus.read(scenario.name)
-  if (fs.existsSync(tmp) && expected && corpus.matches(expected, lines)) {
+  if (fs.existsSync(tmp) && expected && corpus.matchesScenario(scenario, expected, lines)) {
     fs.renameSync(tmp, file)
     console.log(`  native log ${path.relative(process.cwd(), file)}`)
   } else {

test/contract/contract.test.js

@@ -24,7 +24,7 @@ describe('event contract', function () {
       const expected = corpus.read(scenario.name)
       assert.isNotNull(expected, `no transcript for ${scenario.name} - record one with: node test/contract/capture.js --update ${scenario.name}`)
       const lines = await runIsolated(root, connectionString, scenario)
-      if (!corpus.matches(expected, lines)) {
+      if (!corpus.matchesScenario(scenario, expected, lines)) {
         assert.fail(`${scenario.name} (${expected.status}) differs from ${path.basename(expected.file)}:\n${corpus.diff(expected.variants[0], lines)}`)
       }
     })

test/contract/corpus.js

@@ -30,8 +30,8 @@ function fileFor (name, platform = process.platform) {
   return path.join(CORPUS_DIR, `${name}.txt`)
 }
 
-function read (name) {
-  const file = fileFor(name)
+function read (name, platform = process.platform) {
+  const file = fileFor(name, platform)
   if (!fs.existsSync(file)) return null
   const text = fs.readFileSync(file, 'utf8').replace(/\r\n/g, '\n')
   const status = /^# status: (\S+)/m.exec(text)?.[1] ?? 'unreviewed'
@@ -102,4 +102,30 @@ function diff (expected, actual) {
   return out.join('\n')
 }
 
-module.exports = { read, write, matches, diff, CORPUS_DIR }
+// a timed scenario's pause comes from a timer, so how the rows split around it depends on
+// timing: compare its shape (consecutive rows as one line) and the number of rows of each
+// query, not the split
+const ROW = /^((?:q\d+: )?row)(?: x(\d+))?$/
+function shape (lines) {
+  const out = []
+  const rows = {}
+  for (const l of lines) {
+    const m = ROW.exec(l)
+    if (!m) {
+      out.push(l)
+      continue
+    }
+    rows[m[1]] = (rows[m[1]] ?? 0) + Number(m[2] ?? 1)
+    if (out[out.length - 1] !== m[1]) out.push(m[1])
+  }
+  return JSON.stringify([out, rows])
+}
+
+// whether lines are an accepted transcript of the scenario
+function matchesScenario (scenario, expected, lines) {
+  if (!scenario?.timed) return matches(expected, lines)
+  const actual = shape(lines)
+  return expected.variants.some(v => shape(v) === actual)
+}
+
+module.exports = { read, write, matches, matchesScenario, diff, CORPUS_DIR }

test/contract/corpus/pause/before-execute.events.txt

@@ -1,11 +1,11 @@
 # scenario: pause/before-execute.events
-# status: unreviewed
+# status: agreed
 # sql: select 1 as n
 # at: {"start":["pause","resume:50"]}
 action pause
 submitted
-meta n
 action resume
+meta n
 row
 done
 free

test/contract/corpus/pause/during-execute-then-error.events.txt

@@ -0,0 +1,12 @@
+# scenario: pause/during-execute-then-error.events
+# status: unreviewed
+# sql: select 1 as n; waitfor delay '00:00:00.300'; raiserror('boom', 16, 1)
+# at: {"later":{"100":["pause","resume:300"]}}
+submitted
+action pause
+action resume
+meta n
+row
+error 42000 50000 boom more=false
+free
+check ok

test/contract/corpus/pause/during-execute-then-result.events.txt

@@ -0,0 +1,14 @@
+# scenario: pause/during-execute-then-result.events
+# status: agreed
+# sql: select 1 as a; waitfor delay '00:00:00.300'; select 2 as b
+# at: {"later":{"100":["pause","resume:300"]}}
+submitted
+action pause
+action resume
+meta a
+row
+meta b
+row
+done
+free
+check ok

test/contract/corpus/pause/during-fetch.events.txt

@@ -0,0 +1,15 @@
+# scenario: pause/during-fetch.events
+# status: agreed
+# sql: select top 1000 a.name from sys.all_objects a cross join sys.all_objects b; waitfor delay '00:00:00.400'; select 2 as b
+# at: {"later":{"200":["pause","resume:300"]}}
+submitted
+meta name
+row x800
+action pause
+action resume
+row x200
+meta b
+row
+done
+free
+check ok