also looking at this
feat: add bounded TCP transport for autopilot drivers
#71
22 files
+1798
-154
Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.
| @@ -247,11 +247,10 @@ test('a native connect callback after the absolute setup deadline cannot establi | ||
| 247 | 247 | finally { driver?.close(); server.stop(true) } |
| 248 | 248 | }, 6000) |
| 249 | 249 | |
| 250 | async function fillNativeBackpressure(driver: TcpAdapter) { | |
| 251 | const fillDeadline = performance.now() + 10000 | |
| 250 | async function fillNativeBackpressure(driver: TcpAdapter, sendBudget = 550000, fillDeadline = performance.now() + 10000) { | |
| 252 | 251 | let attempted = 0 |
| 253 | while (!driver.transportSnapshot().backpressured && !driver.protocol.snapshot().fault && attempted < 550000 && performance.now() < fillDeadline) { | |
| 254 | for (let index = 0; index < 128 && attempted < 550000 && !driver.transportSnapshot().backpressured && !driver.protocol.snapshot().fault; index++) { | |
| 252 | while (!driver.transportSnapshot().backpressured && !driver.protocol.snapshot().fault && attempted < sendBudget && performance.now() < fillDeadline) { | |
| 253 | for (let index = 0; index < 128 && attempted < sendBudget && !driver.transportSnapshot().backpressured && !driver.protocol.snapshot().fault; index++) { | |
| 255 | 254 | driver.send({ kind: 'hold', reason: 'native drain deadline' }); attempted++ |
| 256 | 255 | } |
| 257 | 256 | await Bun.sleep(0) |
| @@ -259,19 +258,61 @@ async function fillNativeBackpressure(driver: TcpAdapter) { | ||
| 259 | 258 | return { attempted, blocked: driver.transportSnapshot() } |
| 260 | 259 | } |
| 261 | 260 | |
| 261 | // A native pressure phase may make legitimate progress before its deadline. | |
| 262 | // Continue establishment only after observing that transition within its budget. | |
| 263 | async function observeNativeOutputExpiry(driver: TcpAdapter, mode: 'drain' | 'progress') { | |
| 264 | const fillDeadline = performance.now() + 10000 | |
| 265 | let attempted = 0, phases = 0 | |
| 266 | const history: Array<{ before: ReturnType<TcpAdapter['transportSnapshot']>, after: ReturnType<TcpAdapter['transportSnapshot']>, elapsedMs: number, fault: string | null }> = [] | |
| 267 | try { | |
| 268 | while (performance.now() < fillDeadline && attempted < 550000) { | |
| 269 | let before = driver.transportSnapshot() | |
| 270 | if (mode === 'drain') { | |
| 271 | const phase = await fillNativeBackpressure(driver, 550000 - attempted, fillDeadline) | |
| 272 | attempted += phase.attempted; before = phase.blocked | |
| 273 | } | |
| 274 | else if (before.pendingCallbackBytes === 0 && !before.backpressured) { | |
| 275 | const phase = await fillNativePendingWrites(driver, 550000 - attempted, fillDeadline) | |
| 276 | attempted += phase.attempted; before = phase.pending | |
| 277 | } | |
| 278 | const epoch = mode === 'drain' ? before.blockedSinceMs : before.lastWriteProgressAtMs | |
| 279 | const ready = mode === 'drain' ? before.backpressured : !before.backpressured && before.pendingCallbackBytes > 0 | |
| 280 | if (!ready || before.state !== 'connected' || epoch === null) throw new Error(`Native TCP ${mode} pressure was not established within its byte/time budget`) | |
| 281 | const started = performance.now() | |
| 282 | try { | |
| 283 | // No writes while waiting. A continuous stall keeps the original 1500 ms | |
| 284 | // immediate-failure guard; a valid transition starts a new pressure phase. | |
| 285 | await until(() => { | |
| 286 | if (driver.protocol.snapshot().fault !== null) return true | |
| 287 | const after = driver.transportSnapshot() | |
| 288 | return mode === 'drain' ? !after.backpressured : after.pendingCallbackBytes === 0 || after.lastWriteProgressAtMs !== before.lastWriteProgressAtMs | |
| 289 | }) | |
| 290 | } | |
| 291 | finally { | |
| 292 | phases++ | |
| 293 | if (history.length === 16) history.shift() | |
| 294 | history.push({ before, after: driver.transportSnapshot(), elapsedMs: performance.now() - started, fault: driver.protocol.snapshot().fault }) | |
| 295 | } | |
| 296 | if (driver.protocol.snapshot().fault !== null) return { attempted, pressure: before } | |
| 297 | if (performance.now() - epoch >= 500) throw new Error(`Native TCP ${mode} transition was late or timing-ambiguous; pressure establishment cannot continue`) | |
| 298 | } | |
| 299 | throw new Error(`Native TCP sustained ${mode} pressure was not established within its byte/time budget`) | |
| 300 | } | |
| 301 | finally { console.log(JSON.stringify({ campaign: `native-tcp-${mode}-phases`, attempted, phases, retainedPhaseLimit: 16, history })) } | |
| 302 | } | |
| 303 | ||
| 262 | 304 | test('native TCP backpressure expires the drain deadline without an additional write', async () => { |
| 263 | 305 | const remote = await createPressurePeer(telemetry(3)) |
| 264 | 306 | let driver: TcpAdapter | undefined |
| 265 | 307 | try { |
| 266 | 308 | driver = await createArduPilotDriver({ transport: 'tcp', kind: 'sitl', port: remote.port }) |
| 267 | 309 | await until(() => driver!.read()?.batteryPercent === 90) |
| 268 | const { attempted, blocked } = await fillNativeBackpressure(driver) | |
| 310 | const { attempted, pressure: blocked } = await observeNativeOutputExpiry(driver, 'drain') | |
| 269 | 311 | console.log(JSON.stringify({ campaign: 'native-tcp-drain', attempted, blocked, receivedBytes: remote.receivedBytes(), fault: driver.protocol.snapshot().fault })) |
| 270 | 312 | expect(blocked.backpressured).toBe(true) |
| 271 | 313 | expect(blocked.state).toBe('connected') |
| 272 | 314 | expect(blocked.pendingCallbackBytes).toBeLessThanOrEqual(16384) |
| 273 | 315 | expect(remote.receivedBytes()).toBe(0) |
| 274 | await until(() => driver!.protocol.snapshot().fault !== null) | |
| 275 | 316 | expect(driver.protocol.snapshot().fault).toContain('TCP output drain deadline expired') |
| 276 | 317 | expect(performance.now() - blocked.blockedSinceMs!).toBeGreaterThanOrEqual(500) |
| 277 | 318 | expect(driver.read()).toBeUndefined() |
| @@ -328,11 +369,10 @@ test('a native drain within budget settles accepted bytes once and transport sna | ||
| 328 | 369 | finally { driver?.close(); await remote.close() } |
| 329 | 370 | }, 15000) |
| 330 | 371 | |
| 331 | async function fillNativePendingWrites(driver: TcpAdapter) { | |
| 332 | const fillDeadline = performance.now() + 10000 | |
| 372 | async function fillNativePendingWrites(driver: TcpAdapter, sendBudget = 550000, fillDeadline = performance.now() + 10000) { | |
| 333 | 373 | let attempted = 0, observedPending = false |
| 334 | while (!observedPending && !driver.protocol.snapshot().fault && attempted < 550000 && performance.now() < fillDeadline) { | |
| 335 | for (let index = 0; index < 16 && attempted < 550000 && !driver.transportSnapshot().backpressured; index++) { | |
| 374 | while (!observedPending && !driver.protocol.snapshot().fault && attempted < sendBudget && performance.now() < fillDeadline) { | |
| 375 | for (let index = 0; index < 16 && attempted < sendBudget && !driver.transportSnapshot().backpressured; index++) { | |
| 336 | 376 | driver.send({ kind: 'hold', reason: 'native callback progress deadline' }); attempted++ |
| 337 | 377 | } |
| 338 | 378 | await Bun.sleep(0) |
| @@ -350,15 +390,13 @@ test('native TCP writes below the watermark expire when their callbacks make no | ||
| 350 | 390 | try { |
| 351 | 391 | driver = await createArduPilotDriver({ transport: 'tcp', kind: 'sitl', port: remote.port }) |
| 352 | 392 | await until(() => driver!.read()?.batteryPercent === 90) |
| 353 | const { attempted, pending, observedPending } = await fillNativePendingWrites(driver) | |
| 354 | console.log(JSON.stringify({ campaign: 'native-tcp-write-progress', attempted, pending, observedPending, receivedBytes: remote.receivedBytes(), fault: driver.protocol.snapshot().fault })) | |
| 355 | expect(observedPending).toBe(true) | |
| 393 | const { attempted, pressure: pending } = await observeNativeOutputExpiry(driver, 'progress') | |
| 394 | console.log(JSON.stringify({ campaign: 'native-tcp-write-progress', attempted, pending, receivedBytes: remote.receivedBytes(), fault: driver.protocol.snapshot().fault })) | |
| 356 | 395 | expect(pending.state).toBe('connected') |
| 357 | 396 | expect(pending.backpressured).toBe(false) |
| 358 | 397 | expect(pending.pendingCallbackBytes).toBeGreaterThan(0) |
| 359 | 398 | expect(pending.runtimeWritableBytes).toBeLessThan(4096) |
| 360 | 399 | expect(remote.receivedBytes()).toBe(0) |
| 361 | await until(() => driver!.protocol.snapshot().fault !== null) | |
| 362 | 400 | expect(driver.protocol.snapshot().fault).toContain('TCP write progress deadline expired') |
| 363 | 401 | expect(performance.now() - pending.lastWriteProgressAtMs!).toBeGreaterThanOrEqual(500) |
| 364 | 402 | expect(driver.read()).toBeUndefined() |