diff --git a/docs/adr/0007-remote-device-leases.md b/docs/adr/0007-remote-device-leases.md index af0e02d6ac..eaf4b8a6ab 100644 --- a/docs/adr/0007-remote-device-leases.md +++ b/docs/adr/0007-remote-device-leases.md @@ -67,6 +67,42 @@ one minute, while a cloud WebDriver connection profile asks for ten. A single co longer than its own lease is therefore ordinary on the default and only reachable through a profile on the longer one. +## Client-side work that precedes admission + +Protecting admitted work covers nothing that happens before a request is admitted. Installing an +artifact uploads it from the caller while the request that will consume it has not been admitted yet, +so an upload slower than the lease's inactivity window expired the lease paying for the device the +bytes were going to (#2946). The caller therefore beats the lease over the ordinary transport for as +long as that phase runs, naming the lease scope exactly as the command named it and no payload of its +own. An install names no window, so its beats renew for the window the lease already carries; a +caller that did name one keeps renewing on it. Resolving an absent window to the registry default +instead — which is what a heartbeat used to do — quietly shortened every lease allocated above that +default, which is the other half of why the upload could not survive. Request admission had its own +copy of that mistake: it named a proxy-specific default for every admitted request, so a lease +allocated longer than the default lost its window on the next command. Admission renews on the lease's +window too; the window a lease carries is the one its client named when it allocated. + +The first beat is fired when the phase starts, not one cadence in, because a beat is what proves the +lease the upload is spending time on is still alive — a lease shorter than any fixed cadence would +otherwise lapse before anything renewed it. A lease already gone is caught early, but not before the +phase begins: hashing, the preflight, and the start of the stream can all run while the first beat is +still outstanding, so what the first beat buys is that the loss is learned during the upload rather +than after it. Each beat answers with the window it just renewed, and the loop arms its successor when +the beat starts rather than when it settles, so a beat that never answers is abandoned on schedule +instead of taking the schedule with it; the beat's own budget is capped at the cadence it started on, +which is what keeps one stalled round trip from outliving the window it exists to protect. A phase +that settles does not wait on a beat still in flight. + +A beat is a fresh request each time, never the protected request rewritten: a request identity is +what a timed-out beat is canceled under, and beats must not inherit each other's cancellation. A beat +that finds the lease gone, or finds that this request can never renew it — its scope is missing or +belongs to another lease, or the daemon rejects the scope and window the beat itself was built with, +which no successor will ask differently — ends the phase with that error and cancels the upload rather +than finishing bytes against a device nobody owns or a lease that will stop renewing. A beat that fails +for any other reason is reported and survived, because a later beat covers one lost request — including +one that was abandoned at its budget rather than answered, which is what makes that promise hold for a +stalled round trip and not only for one that fails fast. + ## Human control Human-control holds coexist with an open remote session. They belong to `LeaseRegistry` and use diff --git a/packages/contracts/src/__tests__/lease-scope.test.ts b/packages/contracts/src/__tests__/lease-scope.test.ts index daad373eb2..3da34f2be7 100644 --- a/packages/contracts/src/__tests__/lease-scope.test.ts +++ b/packages/contracts/src/__tests__/lease-scope.test.ts @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import { test } from 'vitest'; import { findMissingProxyLeaseFields, + isInactiveLeaseError, leaseScopeFromOptions, leaseScopeFromRequest, leaseScopeToCommandFlags, @@ -215,3 +216,29 @@ test('readLeaseAllocateProviderFlags carries the provider-allocation flags and d ); assert.deepEqual(readLeaseAllocateProviderFlags(undefined), {}); }); + +test('isInactiveLeaseError recognizes the lease being gone, and refuses a request mismatch', () => { + for (const reason of ['LEASE_NOT_FOUND', 'LEASE_EXPIRED', 'LEASE_REVOKED']) { + assert.equal( + isInactiveLeaseError({ code: 'UNAUTHORIZED', details: { reason } }), + true, + `${reason} says the lease is no longer usable`, + ); + } + // A mismatch is about the request that asked, not the lease: the lease may be perfectly alive and + // held by this same client under another request, so a beat must keep renewing on this answer. + assert.equal( + isInactiveLeaseError({ code: 'UNAUTHORIZED', details: { reason: 'LEASE_SCOPE_MISMATCH' } }), + false, + 'a scope mismatch is not the lease being gone', + ); + // Both halves are required: the code alone is any unauthorized call, and the reason alone could + // ride on an error the client has no business acting on. + assert.equal(isInactiveLeaseError({ code: 'UNAUTHORIZED', details: {} }), false); + assert.equal( + isInactiveLeaseError({ code: 'COMMAND_FAILED', details: { reason: 'LEASE_NOT_FOUND' } }), + false, + ); + assert.equal(isInactiveLeaseError(new Error('LEASE_NOT_FOUND')), false); + assert.equal(isInactiveLeaseError(undefined), false); +}); diff --git a/packages/contracts/src/lease-scope.ts b/packages/contracts/src/lease-scope.ts index 57a2b480f9..d03b7cf6d7 100644 --- a/packages/contracts/src/lease-scope.ts +++ b/packages/contracts/src/lease-scope.ts @@ -4,7 +4,6 @@ import type { CloudProviderProfileFields } from './remote-config-fields.ts'; import type { CommandFlags } from './command-flags.ts'; const PROXY_LEASE_PROVIDER = 'proxy'; -export const DEFAULT_PROXY_LEASE_TTL_MS = 300_000; const REQUIRED_PROXY_LEASE_FIELDS = [ 'leaseId', @@ -296,6 +295,32 @@ export function findMissingProxyLeaseFields(scope: LeaseScope): string[] { return REQUIRED_PROXY_LEASE_FIELDS.filter((field) => !scope[field]); } +/** + * Why a lease stopped being ours: it is gone, spent, or taken back. + * + * This is the whole taxonomy of "the lease is no longer usable" as the daemon reports it, and both + * readers ask the same question — a client deciding whether a connection still owns a device, and a + * lease beat deciding whether an upload is still worth finishing. A reason naming a mismatch between + * a request and a lease is not in here: that says something about the request, not the lease. + */ +const INACTIVE_LEASE_REASONS: ReadonlySet = new Set([ + 'LEASE_NOT_FOUND', + 'LEASE_EXPIRED', + 'LEASE_REVOKED', +]); + +/** Whether `error` says the lease is gone rather than merely unavailable to this request. */ +export function isInactiveLeaseError(error: unknown): boolean { + return ( + typeof error === 'object' && + error !== null && + (error as Readonly<{ code?: unknown }>).code === 'UNAUTHORIZED' && + INACTIVE_LEASE_REASONS.has( + (error as Readonly<{ details?: Readonly<{ reason?: unknown }> }>).details?.reason, + ) + ); +} + function leaseScopeToScopedRequest(scope: LeaseScope): LeaseScopedRequestScope { return stripUndefined({ leaseId: scope.leaseId ?? '', diff --git a/scripts/layering/daemon-client-entry.ts b/scripts/layering/daemon-client-entry.ts index 3c40ddce4a..1c7ced3664 100644 --- a/scripts/layering/daemon-client-entry.ts +++ b/scripts/layering/daemon-client-entry.ts @@ -43,6 +43,11 @@ export const DAEMON_CLIENT_ENTRY_EDGES: readonly DaemonClientEntryEdge[] = [ target: 'src/daemon/daemon-request.ts', rationale: REQUEST_RATIONALE, }, + { + file: 'src/daemon-client/daemon-client-lease-beat.ts', + target: 'src/daemon/daemon-request.ts', + rationale: REQUEST_RATIONALE, + }, { file: 'src/daemon-client/daemon-client-progress.ts', target: 'src/daemon/daemon-request.ts', diff --git a/src/__tests__/upload-client-cancellation.test.ts b/src/__tests__/upload-client-cancellation.test.ts new file mode 100644 index 0000000000..0313feff53 --- /dev/null +++ b/src/__tests__/upload-client-cancellation.test.ts @@ -0,0 +1,199 @@ +// Ending an artifact upload when the work it serves is over (#2946). +// +// A beat that finds the device's lease gone aborts the upload protecting that lease. An upload is a +// piped `node:http` request, so the only way to stop bytes already in flight is the request's own +// abort signal — a rejection the caller swallows would keep streaming a full app bundle to a device +// nobody owns. Kept out of `upload-client.test.ts`, which is already over the test-file size +// tripwire and may not grow (docs/agents/testing.md). + +import { afterEach, test } from 'vitest'; +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import fs from 'node:fs'; +import http, { type IncomingMessage, type ServerResponse } from 'node:http'; +import path from 'node:path'; +import { once } from 'node:events'; +import { uploadArtifact } from '../remote/upload-client.ts'; +import { mkdtempForTestSync } from './test-utils/tmp-dir.ts'; + +const TEST_TOKEN = 'agent-device-upload-cancel-token'; +const tempDirs: string[] = []; + +afterEach(async () => { + for (const dir of tempDirs) { + await fs.promises.rm(dir, { recursive: true, force: true }).catch(() => {}); + } + tempDirs.length = 0; +}); + +test('an aborted signal ends a legacy upload mid-stream instead of finishing the bytes', async () => { + // Two megabytes is well past what a paused read lets through: the assertions below are about the + // stream stopping, and a smaller payload keeps that off the CPU in a loaded lane. + const content = Buffer.alloc(2 * 1024 * 1024, 'x'); + const artifactPath = createTempFile('app.apk', content); + const control = new AbortController(); + let sawBytes = 0; + + // Preflight reports itself unsupported so the upload takes the legacy stream, the one path that + // pipes a file at the daemon and keeps going for as long as the daemon drains it. + const server = await startServer(async (req, res) => { + if (req.method === 'POST' && req.url === '/upload/preflight') { + await readRequestBody(req); + res.statusCode = 404; + res.end('not found'); + return; + } + if (req.method === 'POST' && req.url === '/upload') { + req.on('data', (chunk: Buffer) => { + sawBytes += chunk.length; + // The lease dies once bytes are genuinely in flight — not before the request starts. + // Stopping the read applies backpressure so the rest of the file cannot race through + // loopback and make "stopped early" a matter of timing. + req.pause(); + if (!control.signal.aborted) control.abort(); + }); + return; + } + res.statusCode = 404; + res.end('not found'); + }); + + try { + await assert.rejects( + async () => + await uploadArtifact({ + localPath: artifactPath, + baseUrl: server.baseUrl, + token: TEST_TOKEN, + signal: control.signal, + }), + isAbortError, + ); + assert.ok(sawBytes > 0, 'the upload had started streaming before the abort'); + assert.ok(sawBytes < content.length, `the stream stopped early, at ${sawBytes} bytes`); + } finally { + await server.close(); + } +}); + +test('an aborted signal before preflight refuses to ask the daemon for a ticket', async () => { + const artifactPath = createTempFile('app.apk', 'payload'); + const control = new AbortController(); + control.abort(); + const requests: string[] = []; + + const server = await startServer(async (req, res) => { + requests.push(`${req.method} ${req.url}`); + res.statusCode = 404; + res.end('not found'); + }); + + try { + await assert.rejects( + async () => + await uploadArtifact({ + localPath: artifactPath, + baseUrl: server.baseUrl, + token: TEST_TOKEN, + signal: control.signal, + }), + isAbortError, + ); + assert.deepEqual(requests, [], 'an upload nobody waits for never reaches the daemon'); + } finally { + await server.close(); + } +}); + +test('an upload with no signal behaves exactly as before', async () => { + const content = 'unprotected-payload'; + const artifactPath = createTempFile('app.apk', content); + const expectedHash = createHash('sha256').update(content).digest('hex'); + + const server = await startServer(async (req, res) => { + if (req.method === 'POST' && req.url === '/upload/preflight') { + await readRequestBody(req); + res.statusCode = 404; + res.end('not found'); + return; + } + if (req.method === 'POST' && req.url === '/upload') { + assert.equal(req.headers['x-artifact-hash'], expectedHash); + await readRequestBody(req); + sendJson(res, { ok: true, uploadId: 'upload-uncancelled' }); + return; + } + res.statusCode = 404; + res.end('not found'); + }); + + try { + const uploadId = await uploadArtifact({ + localPath: artifactPath, + baseUrl: server.baseUrl, + token: TEST_TOKEN, + }); + assert.equal(uploadId, 'upload-uncancelled'); + } finally { + await server.close(); + } +}); + +function isAbortError(error: unknown): boolean { + // The contract on `signal` is that an aborted upload rejects with the signal's own reason. Any + // other rejection — a transport failure, the server-error fallback — means the upload stopped for + // a reason this test is not about and would let the abort path regress while staying green. + return error instanceof DOMException && error.name === 'AbortError'; +} + +function createTempFile(filename: string, content: string | Buffer): string { + const dir = mkdtempForTestSync('agent-device-upload-cancel-'); + tempDirs.push(dir); + const filePath = path.join(dir, filename); + fs.writeFileSync(filePath, content); + return filePath; +} + +async function startServer( + handler: (req: IncomingMessage, res: ServerResponse) => Promise, +): Promise<{ baseUrl: string; close: () => Promise }> { + const server = http.createServer((req, res) => { + void handler(req, res).catch(() => { + res.statusCode = 500; + res.end(); + }); + }); + // A canceled upload leaves its socket in a state `closeAllConnections` can race; without a bound + // idle keep-alive, `close` would then wait out Node's five-second default. + server.keepAliveTimeout = 100; + server.listen(0, '127.0.0.1'); + server.unref(); + await once(server, 'listening'); + const address = server.address(); + assert.ok(address && typeof address === 'object'); + return { + baseUrl: `http://127.0.0.1:${address.port}`, + close: async () => { + // A canceled upload leaves its socket half-open and a pooled keep-alive one behind; without + // dropping them, `close` would wait out the server's five-second keep-alive timeout. + server.closeAllConnections(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + }, + }; +} + +async function readRequestBody(req: IncomingMessage): Promise { + const chunks: Buffer[] = []; + for await (const chunk of req) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + return Buffer.concat(chunks); +} + +function sendJson(res: ServerResponse, body: unknown): void { + res.statusCode = 200; + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify(body)); +} diff --git a/src/cli/commands/connection-runtime.ts b/src/cli/commands/connection-runtime.ts index 961560e226..23579c540d 100644 --- a/src/cli/commands/connection-runtime.ts +++ b/src/cli/commands/connection-runtime.ts @@ -15,6 +15,7 @@ import { type DeviceInfo, } from '@agent-device/kernel/device'; import { shouldAgentCdpUseRemoteBridgeUrl } from './agent-cdp.ts'; +import { isInactiveLeaseError } from '@agent-device/contracts/lease-scope'; import { buildRemoteConnectionDaemonState, buildRemoteConnectionRequestMetadata, @@ -991,12 +992,3 @@ async function heartbeatOrAllocateLease( throw error; } } - -function isInactiveLeaseError(error: unknown): boolean { - if (!(error instanceof AppError) || error.code !== 'UNAUTHORIZED') return false; - return ( - error.details?.reason === 'LEASE_NOT_FOUND' || - error.details?.reason === 'LEASE_EXPIRED' || - error.details?.reason === 'LEASE_REVOKED' - ); -} diff --git a/src/daemon-client/__tests__/daemon-client-lease-beat.test.ts b/src/daemon-client/__tests__/daemon-client-lease-beat.test.ts new file mode 100644 index 0000000000..cc6551c274 --- /dev/null +++ b/src/daemon-client/__tests__/daemon-client-lease-beat.test.ts @@ -0,0 +1,732 @@ +import { afterEach, describe, test, vi } from 'vitest'; +import assert from 'node:assert/strict'; +import net from 'node:net'; +import { AppError } from '@agent-device/kernel/errors'; +import { closeLoopbackServer, listenOnLoopback } from '../../__tests__/test-utils/loopback.ts'; +import { resolveDaemonPaths } from '../../daemon-resolution.ts'; +import { + buildLeaseHeartbeatRequest, + buildUploadLeaseHeartbeat, + createLeaseRenewalBeat, + leaseScopeForHeartbeat, + runProtectedLeaseWork, +} from '../daemon-client-lease-beat.ts'; +import type { DaemonRequest } from '../../daemon/daemon-request.ts'; + +function lostLeaseError(reason: string): AppError { + return new AppError('UNAUTHORIZED', 'Lease is not active', { reason }); +} + +function deferred(): { + promise: Promise; + resolve: (value: T) => void; + reject: (error: unknown) => void; +} { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +afterEach(() => { + vi.useRealTimers(); +}); + +function renewedLeaseResponse(windowMs: number): { + ok: true; + data: { lease: { heartbeatAt: number; expiresAt: number } }; +} { + return { ok: true, data: { lease: { heartbeatAt: 1_000_000, expiresAt: 1_000_000 + windowMs } } }; +} + +describe('runProtectedLeaseWork', () => { + test('runs the task untouched when there is no lease to protect', async () => { + const heartbeat = vi.fn(); + const phase = await runProtectedLeaseWork({ heartbeat: undefined, task: async () => 'ok' }); + assert.equal(phase, 'ok'); + assert.equal(heartbeat.mock.calls.length, 0); + }); + + test('a fast upload that lands before the first beat reports success', async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => renewedLeaseResponse(30_000)); + const running = runProtectedLeaseWork({ + task: async (signal) => { + assert.ok(!signal.aborted, 'a lease still held does not cancel the upload'); + return 'installed'; + }, + heartbeat, + }); + + assert.equal(await running, 'installed', 'the beat is armed but the upload is faster'); + }); + + test('the first beat fires immediately, not one interval into the upload', async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => renewedLeaseResponse(30_000)); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + // A lease admitted with a short window used to be able to lapse before the first beat: the + // upload paid for the device with a lease nobody renewed yet. + await vi.advanceTimersByTimeAsync(0); + assert.equal(heartbeat.mock.calls.length, 1, 'a beat is out before any interval elapses'); + + upload.resolve('installed'); + assert.equal(await running, 'installed'); + }); + + test('beats follow the window the lease reports, not the assumed floor', async () => { + vi.useFakeTimers(); + // The daemon says it just extended the lease by 15s; the next beat must land a third of that + // after the answer, whatever the loop assumed beforehand. + const heartbeat = vi.fn(async () => renewedLeaseResponse(15_000)); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(5_000); + assert.equal(heartbeat.mock.calls.length, 2, 'first beat at once, second a third of 15s in'); + await vi.advanceTimersByTimeAsync(5_000); + assert.equal(heartbeat.mock.calls.length, 3); + + upload.resolve('installed'); + assert.equal(await running, 'installed'); + const before = heartbeat.mock.calls.length; + await vi.advanceTimersByTimeAsync(60_000); + assert.equal(heartbeat.mock.calls.length, before, 'no beat outlives the phase'); + }); + + test('a beat shorter than the floor still beats no faster than the floor', async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => renewedLeaseResponse(1_500)); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(999); + assert.equal(heartbeat.mock.calls.length, 1, 'the floor holds a pathological window off'); + await vi.advanceTimersByTimeAsync(1); + assert.equal(heartbeat.mock.calls.length, 2); + + upload.resolve('installed'); + assert.equal(await running, 'installed'); + }); + + test('an answer that names no window keeps the cadence the beat was asked at', async () => { + vi.useFakeTimers(); + // Slowing down here would have to rest on evidence about the lease, and an unreadable answer is + // the absence of evidence: an older daemon renews happily without describing the window. + const heartbeat = vi.fn(async () => ({ ok: true, data: {} })); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(1_000); + assert.equal(heartbeat.mock.calls.length, 2, 'immediate first, then the same cadence again'); + await vi.advanceTimersByTimeAsync(1_000); + assert.equal(heartbeat.mock.calls.length, 3); + + upload.resolve('installed'); + assert.equal(await running, 'installed'); + }); + + test('each beat is given a budget no longer than the cadence it starts on', async () => { + vi.useFakeTimers(); + const budgets: number[] = []; + const heartbeat = vi.fn(async (budgetMs: number) => { + budgets.push(budgetMs); + return renewedLeaseResponse(60_000); + }); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(40_000); + // Before the first answer the loop assumes the shortest window the daemon accepts, so the beat + // that has to prove that lease is alive cannot itself take longer than a fifth of it. Once the + // window is known the budget is the cadence: a stalled beat dies inside one, not the 90s the + // command's own heartbeat policy would allow. + assert.deepEqual(budgets, [1_000, 20_000, 20_000]); + + upload.resolve('installed'); + assert.equal(await running, 'installed'); + }); + + test('a beat that never settles is abandoned on schedule and holds no successor off', async () => { + vi.useFakeTimers(); + // The #2946 failure this loop exists to prevent, in its remaining shape: a beat stuck on a + // half-open connection. Overlapping is the point — a stalled beat must not take the schedule + // with it, or the lease dies at 60s while the beat waits out its transport timeout. + const started: number[] = []; + const stalled = deferred(); + const heartbeat = vi.fn(async () => { + started.push(Date.now()); + await stalled.promise; + return renewedLeaseResponse(5_000); + }); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + // A lease on the registry's five-second minimum window, with every beat stalled. + await vi.advanceTimersByTimeAsync(4_000); + assert.ok( + heartbeat.mock.calls.length >= 3, + `beats keep coming through a stall, got ${heartbeat.mock.calls.length}`, + ); + + upload.resolve('installed'); + let outcome: string | undefined; + void running.then((value) => { + outcome = value; + }); + await vi.advanceTimersByTimeAsync(0); + assert.equal(outcome, 'installed', 'a finished upload does not wait on a stalled beat'); + }); + + test('a late lost-lease answer from an abandoned beat is logged and does not rewrite the outcome', async () => { + vi.useFakeTimers(); + let reportLost: ((error: unknown) => void) | undefined; + const heartbeat = vi.fn( + () => + new Promise((_resolve, reject) => { + reportLost = reject; + }), + ); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(3_000); + assert.equal(heartbeat.mock.calls.length, 4, 'the stalled beats were abandoned, not awaited'); + + upload.resolve('installed'); + let outcome: string | undefined; + void running.then( + () => { + outcome = 'resolved'; + }, + (error: unknown) => { + outcome = error instanceof AppError ? String(error.details?.reason) : 'rejected'; + }, + ); + await vi.advanceTimersByTimeAsync(0); + assert.equal(outcome, 'resolved', 'the upload had already finished when the beat was dropped'); + + // The abandoned beat is still listened to: its late answer is a fact about the lease the caller + // should be able to find in the log even though its own outcome already stands. + reportLost?.(lostLeaseError('LEASE_EXPIRED')); + await vi.advanceTimersByTimeAsync(0); + assert.equal(outcome, 'resolved'); + }); + + test('a beat that fails for a transient reason is survived and re-armed', async () => { + vi.useFakeTimers(); + // A reason from the same registry that is not a lost lease: contention says nothing about + // whether this lease is still ours, so the upload keeps going and the next beat asks again. + const heartbeat = vi.fn<() => Promise>(async () => { + throw new AppError('DEVICE_IN_USE', 'Device is already leased', { + reason: 'DEVICE_LEASE_BUSY', + }); + }); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(2_500); + assert.equal(heartbeat.mock.calls.length, 3, 'one failed beat does not stop the others'); + + heartbeat.mockImplementation(async () => renewedLeaseResponse(30_000)); + await vi.advanceTimersByTimeAsync(1_000); + upload.resolve('installed'); + assert.equal(await running, 'installed'); + }); + + for (const reason of ['LEASE_NOT_FOUND', 'LEASE_EXPIRED', 'LEASE_REVOKED']) { + test(`a beat that finds the lease ${reason} ends the phase with that error`, async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => { + throw lostLeaseError(reason); + }); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + // #2946 asked for an upload longer than the TTL to still succeed; when the lease really is + // gone the honest answer is the lease error, delivered before the bytes finish. The + // expectation rides the promise before the clock moves, so the rejection is never unobserved. + const rejected = assert.rejects( + running, + (error: unknown) => + error instanceof AppError && + error.code === 'UNAUTHORIZED' && + error.details?.reason === reason, + ); + await vi.advanceTimersByTimeAsync(0); + await rejected; + upload.resolve('too late'); + }); + } + + for (const reason of ['LEASE_SCOPE_REQUIRED', 'LEASE_SCOPE_MISMATCH']) { + test(`a beat refused ${reason} ends the phase instead of beating to the lease's death`, async () => { + vi.useFakeTimers(); + // Both say this request can never renew the lease — the scope it names is missing or belongs + // to someone else. Surviving would spend the whole upload on a lease that stops renewing: + // the #2946 symptom recreated on the client's own side. + const heartbeat = vi.fn(async () => { + throw new AppError('UNAUTHORIZED', "Lease scope is not this request's", { reason }); + }); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + const rejected = assert.rejects( + running, + (error: unknown) => + error instanceof AppError && + error.code === 'UNAUTHORIZED' && + error.details?.reason === reason, + ); + await vi.advanceTimersByTimeAsync(5_000); + assert.equal( + heartbeat.mock.calls.length, + 1, + 'a doomed renewal is not retried for the window', + ); + await rejected; + upload.resolve('too late'); + }); + } + + test('a beat refused INVALID_ARGS ends the phase, because the beat asks the same thing forever', async () => { + vi.useFakeTimers(); + // The beat's scope and ttl are fixed when it is built, so a daemon that rejects them — a ttl + // outside [minLeaseTtlMs, maxLeaseTtlMs] — rejects every successor identically. It carries no + // reason to key on, so the code is the signal; waiting it out only spends the upload. + const heartbeat = vi.fn(async () => { + throw new AppError('INVALID_ARGS', 'Lease ttlMs must be between 5000 and 3600000.'); + }); + const upload = deferred(); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + const rejected = assert.rejects( + running, + (error: unknown) => error instanceof AppError && error.code === 'INVALID_ARGS', + ); + await vi.advanceTimersByTimeAsync(5_000); + assert.equal(heartbeat.mock.calls.length, 1, 'a refusal the beat cannot fix is not retried'); + await rejected; + upload.resolve('too late'); + }); + + test('a beat that ends the protection cancels the upload the phase is running', async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => { + throw lostLeaseError('LEASE_NOT_FOUND'); + }); + let sawAbort = false; + const running = runProtectedLeaseWork({ + task: (signal) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { + sawAbort = true; + reject(signal.reason); + }); + }), + heartbeat, + }); + + const rejected = assert.rejects(running); + await vi.advanceTimersByTimeAsync(0); + await rejected; + assert.equal(sawAbort, true, 'the upload is told to stop before the bytes finish'); + }); + + test('a phase that throws synchronously still stops the beats', async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => ({ ok: true })); + await assert.rejects( + (async () => + await runProtectedLeaseWork({ + task: () => { + throw new AppError('INVALID_ARGS', 'artifact vanished'); + }, + heartbeat, + }))(), + /artifact vanished/, + ); + + await vi.advanceTimersByTimeAsync(10_000); + assert.equal(heartbeat.mock.calls.length, 0, 'no timer outlived a phase that never started'); + }); + + test('a lost lease surfaces as a rejection even when the task finishes first', async () => { + vi.useFakeTimers(); + + let beat: (() => void) | undefined; + const first = deferred(); + const running = runProtectedLeaseWork({ + task: () => first.promise, + heartbeat: () => + new Promise((_, reject) => { + beat = () => reject(lostLeaseError('LEASE_NOT_FOUND')); + }), + }); + + const rejected = assert.rejects( + running, + (error: unknown) => error instanceof AppError && error.details?.reason === 'LEASE_NOT_FOUND', + ); + await vi.advanceTimersByTimeAsync(0); + assert.ok(beat, 'a beat started'); + first.resolve('installed'); + beat?.(); + await vi.advanceTimersByTimeAsync(0); + await rejected; + }); + + test('a task rejection propagates and still stops the beats', async () => { + vi.useFakeTimers(); + const heartbeat = vi.fn(async () => ({ ok: true })); + const running = runProtectedLeaseWork({ + task: async () => { + throw new AppError('COMMAND_FAILED', 'upload failed'); + }, + heartbeat, + }); + + await assert.rejects((async () => await running)(), /upload failed/); + const before = heartbeat.mock.calls.length; + await vi.advanceTimersByTimeAsync(10_000); + assert.equal(heartbeat.mock.calls.length, before); + }); + + test('an abandoned beat arms no successor once the phase has settled', async () => { + vi.useFakeTimers(); + const beat = deferred(); + const upload = deferred(); + const heartbeat = vi.fn(async () => { + await beat.promise; + return renewedLeaseResponse(30_000); + }); + const running = runProtectedLeaseWork({ + task: () => upload.promise, + heartbeat, + }); + + await vi.advanceTimersByTimeAsync(0); + assert.equal(heartbeat.mock.calls.length, 1, 'a beat is in flight while the upload finishes'); + + upload.resolve('installed'); + assert.equal(await running, 'installed', 'the phase does not wait on a beat it gave up on'); + + beat.resolve(); + await vi.advanceTimersByTimeAsync(120_000); + assert.equal(heartbeat.mock.calls.length, 1, 'the renewal that landed last arms no successor'); + }); +}); + +describe('leaseScopeForHeartbeat', () => { + test('names no lease for a request that carries none', () => { + assert.equal(leaseScopeForHeartbeat({ flags: {}, meta: undefined }), undefined); + }); + + test('reads the lease scope from the request meta, then from the flags', () => { + assert.equal( + leaseScopeForHeartbeat({ meta: { leaseId: 'lease-meta', tenantId: 'acme' } })?.leaseId, + 'lease-meta', + ); + assert.equal( + leaseScopeForHeartbeat({ flags: { leaseId: 'lease-flag' } })?.leaseId, + 'lease-flag', + ); + }); +}); + +describe('buildLeaseHeartbeatRequest', () => { + test('carries the lease scope and nothing that belongs to the request it is protecting', () => { + const beat = buildLeaseHeartbeatRequest( + { + leaseId: 'lease-1', + tenantId: 'acme', + runId: 'run-1', + leaseBackend: 'android-instance', + leaseProvider: 'proxy', + deviceKey: 'android:mobile:emulator-5554', + clientId: 'client-1', + }, + { + session: 'adc-android', + sessionIsolation: 'tenant', + requestId: 'beat-1', + token: 'daemon-token', + }, + ); + + assert.equal(beat.command, 'lease_heartbeat'); + assert.deepEqual(beat.positionals, []); + assert.equal(beat.session, 'adc-android'); + assert.equal(beat.token, 'daemon-token'); + assert.deepEqual(beat.meta, { + leaseId: 'lease-1', + tenantId: 'acme', + runId: 'run-1', + leaseBackend: 'android-instance', + leaseProvider: 'proxy', + deviceKey: 'android:mobile:emulator-5554', + clientId: 'client-1', + sessionIsolation: 'tenant', + requestId: 'beat-1', + }); + // The socket transport serializes the whole request, so a beat that rode along with the install + // would re-send a 449 MB artifact every interval. + assert.equal(beat.flags, undefined); + assert.equal('installSource' in (beat.meta ?? {}), false); + assert.equal(beat.internal, undefined); + }); + + test('sends no ttl for an install, whose scope never carried one, so the lease keeps its own window', () => { + // connection-runtime passes the TTL to `leases.allocate` only, so an install request's scope has + // none. A beat that invented one would shorten a lease allocated longer. + const installRequest: Pick = { + flags: { leaseId: 'lease-1', platform: 'android' }, + meta: { leaseId: 'lease-1', tenantId: 'acme' }, + }; + const scope = leaseScopeForHeartbeat(installRequest)!; + const beat = buildLeaseHeartbeatRequest(scope, { + session: 'default', + requestId: 'beat-1', + token: 't', + }); + assert.equal(beat.meta?.leaseTtlMs, undefined); + }); + + test('a caller that did name a ttl keeps renewing on it', () => { + const scope = leaseScopeForHeartbeat({ + flags: { leaseId: 'lease-1' }, + meta: { leaseId: 'lease-1', leaseTtlMs: 600_000 }, + })!; + const beat = buildLeaseHeartbeatRequest(scope, { + session: 'default', + requestId: 'beat-1', + token: 't', + }); + assert.equal(beat.meta?.leaseTtlMs, 600_000); + }); +}); + +describe('createLeaseRenewalBeat', () => { + const scope = { + leaseId: 'lease-1', + tenantId: 'acme', + runId: 'run-1', + leaseBackend: 'android-instance', + } as const; + + function beatContext(send: (request: DaemonRequest, budgetMs: number) => Promise) { + return { + session: 'adc-android', + sessionIsolation: 'tenant' as const, + token: 'daemon-token', + send, + }; + } + + test('sends one lease_heartbeat naming the lease it is protecting', async () => { + const sent: DaemonRequest[] = []; + await createLeaseRenewalBeat( + scope, + beatContext(async (request) => void sent.push(request)), + )(1_000); + + assert.equal(sent.length, 1); + assert.equal(sent[0]!.command, 'lease_heartbeat'); + assert.equal(sent[0]!.meta?.leaseId, 'lease-1'); + assert.equal(sent[0]!.meta?.tenantId, 'acme'); + assert.equal(sent[0]!.meta?.runId, 'run-1'); + assert.equal(sent[0]!.meta?.sessionIsolation, 'tenant'); + assert.equal(sent[0]!.session, 'adc-android'); + assert.equal(sent[0]!.token, 'daemon-token'); + }); + + test('every beat is a distinct request, so a timed-out beat cannot cancel the next one', async () => { + const sent: DaemonRequest[] = []; + const beat = createLeaseRenewalBeat( + scope, + beatContext(async (request) => void sent.push(request)), + ); + await beat(1_000); + await beat(1_000); + await beat(1_000); + + const ids = sent.map((request) => request.meta?.requestId); + assert.equal(new Set(ids).size, 3, 'three beats, three request ids'); + assert.ok(ids.every((id) => typeof id === 'string' && id.length > 0)); + }); + + test('a transport failure propagates to the caller that survives it', async () => { + const beat = createLeaseRenewalBeat( + scope, + beatContext(async () => { + throw new AppError('COMMAND_FAILED', 'connection reset'); + }), + ); + await assert.rejects((async () => await beat(1_000))(), /connection reset/); + }); +}); + +describe('buildUploadLeaseHeartbeat', () => { + const installRequest = { + command: 'install', + positionals: ['/tmp/app.apk'], + session: 'adc-android', + flags: { + leaseId: 'lease-1', + tenantId: 'acme', + runId: 'run-1', + deviceKey: 'android:mobile:emulator-5554', + platform: 'android' as const, + }, + meta: { leaseId: 'lease-1', tenantId: 'acme', runId: 'run-1' }, + }; + + const settings = { + paths: resolveDaemonPaths('/tmp/agent-device-upload-lease'), + transportPreference: 'socket' as const, + serverMode: 'socket' as const, + }; + + test('no beat for a remote request that names no lease, which has nothing to renew', () => { + // The lease-less path is the one an unleased install takes; a timer there would beat a lease + // that does not exist and keep a request alive that owns no device. + assert.equal( + buildUploadLeaseHeartbeat( + { baseUrl: 'http://remote.example.test/agent-device', token: 't', pid: 1 }, + settings, + { ...installRequest, flags: {}, meta: undefined }, + ), + undefined, + ); + }); + + test('no beat for a local daemon, which never uploads and holds no billed device', () => { + assert.equal( + buildUploadLeaseHeartbeat( + { port: 1, token: 't', pid: process.pid }, + settings, + installRequest, + ), + undefined, + ); + }); + + test('the beat reaches a remote daemon over its HTTP endpoint', async () => { + const requests: { method?: string; path?: string; body: string }[] = []; + const server = net.createServer((socket) => { + let body = ''; + socket.on('data', (chunk) => { + body += chunk.toString('utf8'); + const headerEnd = body.indexOf('\r\n\r\n'); + if (headerEnd < 0) return; + const head = body.slice(0, headerEnd); + const [requestLine] = head.split('\r\n'); + const [method, path] = requestLine?.split(' ') ?? []; + requests.push({ method, path, body: body.slice(headerEnd + 4) }); + const payload = JSON.stringify({ + jsonrpc: '2.0', + id: 'x', + result: { ok: true }, + }); + socket.write( + `HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: ${String( + payload.length, + )}\r\n\r\n${payload}`, + ); + socket.end(); + }); + }); + const port = await listenOnLoopback(server); + + try { + const beat = buildUploadLeaseHeartbeat( + { baseUrl: `http://127.0.0.1:${String(port)}/agent-device`, token: 'remote-token', pid: 1 }, + { ...settings, transportPreference: 'auto' }, + installRequest, + ); + assert.ok(beat); + await beat!(5_000); + } finally { + await closeLoopbackServer(server); + } + + const posted = requests.find((request) => request.method === 'POST'); + assert.ok(posted, 'the beat POSTs to the remote daemon'); + const payload = JSON.parse(posted!.body) as { + method: string; + params: Record; + }; + assert.equal(posted!.path, '/agent-device/rpc'); + assert.equal(payload.method, 'agent_device.lease.heartbeat'); + assert.equal(payload.params.leaseId, 'lease-1'); + assert.equal(payload.params.tenantId, 'acme'); + assert.equal(payload.params.runId, 'run-1'); + assert.equal(payload.params.deviceKey, 'android:mobile:emulator-5554'); + + // A beat asks for the same lease to keep going: it names no window, so the daemon renews the one + // the lease already carries. + assert.equal('ttlMs' in payload.params, false); + }); + + test('a beat that never answers dies at its budget, not at the heartbeat policy', async () => { + vi.useFakeTimers(); + // The daemon accepts the connection and never answers — a half-open pipe. The beat exists to + // notice the lease inside its window, so the transport must cut it at the budget the loop + // handed it (here well under the command's 90s lease_heartbeat policy) and destroy the + // socket, instead of holding one half-open for the full policy. + const server = net.createServer(() => {}); + const port = await listenOnLoopback(server); + try { + const beat = buildUploadLeaseHeartbeat( + { baseUrl: `http://127.0.0.1:${String(port)}/agent-device`, token: 'remote-token', pid: 1 }, + { ...settings, transportPreference: 'auto' }, + installRequest, + ); + assert.ok(beat); + const rejected = assert.rejects((async () => await beat!(1_000))(), /timed out/i); + await vi.advanceTimersByTimeAsync(1_000); + await rejected; + } finally { + vi.useRealTimers(); + await closeLoopbackServer(server); + } + }); +}); diff --git a/src/daemon-client/daemon-client-lease-beat.ts b/src/daemon-client/daemon-client-lease-beat.ts new file mode 100644 index 0000000000..992edfaac7 --- /dev/null +++ b/src/daemon-client/daemon-client-lease-beat.ts @@ -0,0 +1,329 @@ +import type { DaemonRequest } from '../daemon/daemon-request.ts'; +import { AppError } from '@agent-device/kernel/errors'; +import { createRequestId, emitDiagnostic } from '@agent-device/host-kit/diagnostics'; +import { INTERNAL_COMMANDS } from '@agent-device/command-registry/catalog'; +import { resolveCommandTimeoutPolicy } from '@agent-device/command-registry/registry'; +import { resolveCommandRequestTimeoutMs } from '@agent-device/command-registry/timeout-policy'; +import { + isInactiveLeaseError, + leaseScopeFromRequest, + leaseScopeToRequestMeta, + type LeaseScope, +} from '@agent-device/contracts/lease-scope'; +import type { DaemonClientSettings } from './daemon-client-lifecycle.ts'; +import { isRemoteDaemon, type DaemonInfo } from './daemon-client-metadata.ts'; +import { sendRequest } from './daemon-client-transport.ts'; + +// The lease beat: what keeps a remote lease alive across a client-side phase the daemon never sees. +// +// A lease renews when a request is admitted, and the daemon protects a lease while admitted work runs +// on it (#2509, ADR 0007). An artifact upload is the mirror image: it runs on the caller's side, +// before the install that consumes it is admitted, so nothing renews the lease while it runs and a +// large enough artifact expired the lease that was paying for the device it was being uploaded to +// (#2946). This module owns the answer — a beat with a budget that ends before the window does — and +// knows nothing about transports beyond the `send` it is handed. + +/** + * The fastest cadence a phase beats at, and the budget each beat gets until the window is known. + * + * Before the first beat answers, the window is unknown and the worst legal case is the registry's + * five-second minimum: a beat that stalls must be abandoned and retried inside that window, or the + * thing the beat exists to prevent happens while it waits. One second is a fifth of that minimum, + * and no window-derived cadence is ever allowed below it, so it is also the loop's floor: a + * misreported or pathologically short window cannot turn the beat into a request loop faster than + * this. + */ +const MIN_LEASE_BEAT_INTERVAL_MS = 1_000; + +/** + * Why a beat stopped protecting the lease even though the lease may live on: this client's request + * will never be the one that renews it. + * + * A beat refused for a missing or mismatched owner scope is a fact about the request, not the + * lease, so every successor is refused identically. Surviving it would only spend the upload against + * a lease that stops renewing — the #2946 failure with extra steps. + */ +const UNRENEWABLE_LEASE_BEAT_REASONS: ReadonlySet = new Set([ + 'LEASE_SCOPE_REQUIRED', + 'LEASE_SCOPE_MISMATCH', +]); + +/** + * Whether a beat failed for a reason every successor will repeat: the lease is gone (the shared + * taxonomy), the daemon refused a fact baked into the beat itself, or this request can never renew + * it. A beat's scope and ttl never change across the phase, so an `INVALID_ARGS` refusal — an + * out-of-range ttl, an unusable lease id — is terminal without waiting out a window it can no + * longer renew. + * + * `LEASE_SESSION_MISMATCH` is deliberately absent: only request admission raises it, and + * `lease_heartbeat` is admission-exempt, so a beat can never receive it. + */ +function isTerminalLeaseBeatError(error: unknown): boolean { + return ( + isInactiveLeaseError(error) || + (error instanceof AppError && + (UNRENEWABLE_LEASE_BEAT_REASONS.has(error.details?.reason) || error.code === 'INVALID_ARGS')) + ); +} + +/** + * Runs one client-side phase under a lease it does not own the clock of. + * + * A lease renews when a request is admitted, and the daemon protects a lease while ADMITTED work + * runs on it (#2509, ADR 0007). An artifact upload is the mirror image: it happens on the caller's + * side, before the install request that will consume it is admitted, so nothing renews the lease + * while it runs and a large enough artifact expired the lease that was paying for the device it was + * being uploaded to (#2946). + * + * `heartbeat` is the caller's transport decision; `undefined` means there is no lease to protect and + * the phase runs untouched. The first beat is fired immediately rather than one interval in, so a + * lease shorter than that interval is renewed before it can lapse. A lease already gone is caught + * early rather than at the end of the phase, though not before the phase begins: hashing, preflight, + * and the first bytes can all happen while the opening beat is still outstanding. Each beat answers + * with the window it just renewed, and the cadence becomes a third of that window. + * + * A beat's budget is the cadence it started on, and its successor is armed when the beat starts + * rather than when it settles. A beat that never returns — a half-open connection, a daemon wedged + * before it admits anything — is therefore abandoned on schedule instead of holding the schedule: + * the lease is still beaten at window/3, and the abandoned round trip is cut off by its own budget + * in the transport rather than by the command's 90-second heartbeat policy. An abandoned beat is + * still listened to, because the answer it eventually gives can be a lost lease. + * + * A beat that fails for a reason that says nothing about this lease is reported and survived — one + * lost request must not fail an upload that a later beat will cover. A beat that finds the lease + * gone, or finds this client can never renew it, ends the phase with that error and aborts the + * signal the phase runs under: the device is no longer ours (or was never reachable through this + * request), and the only honest outcome is to say so before the bytes finish. + */ +export async function runProtectedLeaseWork( + options: Readonly<{ + /** + * One renewal. `budgetMs` is how long this beat may take before the loop abandons it. Absent + * when the request names no lease to renew, which is the ordinary unleased install. + */ + heartbeat?: ((budgetMs: number) => Promise) | undefined; + task: (signal: AbortSignal) => Promise; + }>, +): Promise { + const { heartbeat } = options; + if (!heartbeat) return await options.task(new AbortController().signal); + + const control = new AbortController(); + // Until a beat names the window, the loop assumes the shortest window the daemon will accept: the + // budget of the beat that has to prove a short lease is alive cannot itself be longer than it. + let intervalMs = MIN_LEASE_BEAT_INTERVAL_MS; + let timer: ReturnType | undefined; + let stopped = false; + let terminalError: unknown; + let reportTerminal: ((error: unknown) => void) | undefined; + const terminal = new Promise((_, reject) => { + reportTerminal = reject; + }); + + const runBeat = (): void => { + // Armed while this beat is still outstanding: a beat that never settles is abandoned on + // schedule rather than taking the schedule with it. The beat may re-arm it on the way out. + const arm = (delayMs: number): void => { + if (stopped) return; + if (timer) clearTimeout(timer); + timer = setTimeout(runBeat, delayMs); + }; + arm(intervalMs); + const settle = (async () => { + const budgetMs = intervalMs; + try { + const renewed = leaseWindowFromHeartbeatResponse(await heartbeat(budgetMs)); + // An answer that names no window keeps the cadence it was asked at: the loop only ever + // slows down on evidence of how long the lease is good for, and never on the absence of it. + if (renewed === undefined) return; + const cadence = Math.max(MIN_LEASE_BEAT_INTERVAL_MS, Math.floor(renewed / 3)); + if (cadence === intervalMs) return; + intervalMs = cadence; + // The window just moved, so the next beat is due one cadence from this answer. + arm(cadence); + } catch (error) { + if (isTerminalLeaseBeatError(error)) { + terminalError = error; + if (stopped) { + // The phase settled first; the outcome it returned already stands, but a lease this + // client just learned is gone is worth one diagnostic on the way out. + emitDiagnostic({ + level: 'warn', + phase: 'lease_lost_after_phase', + data: { message: error instanceof Error ? error.message : String(error) }, + }); + return; + } + // The upload is the only thing still consuming this phase's time, and it is pointed at a + // device this client can no longer renew. Stop it rather than finish bytes nobody owns. + control.abort(); + reportTerminal?.(error); + return; + } + emitDiagnostic({ + level: 'warn', + phase: 'lease_heartbeat_failed', + data: { message: error instanceof Error ? error.message : String(error) }, + }); + } + })(); + // A beat the loop has moved on from is still listened to, and nothing awaits it: its outcome is + // swallowed here so a beat nobody is waiting on cannot surface as an unhandled rejection. + void settle.catch(() => undefined); + }; + + // Armed before the phase starts, not one interval in: a beat is what proves the lease the upload + // is spending its time on is still alive. + timer = setTimeout(runBeat, 0); + // A beat already in flight when the phase settles can still report a lost lease; the caller reads + // it from `terminalError`, so nothing may be left racing on this rejection by then. + void terminal.catch(() => undefined); + const phase = await captureOutcome( + // The async boundary also turns a synchronous throw from the phase into a rejection, so the + // timer below is always cleared. + (async () => await Promise.race([options.task(control.signal), terminal]))(), + ); + stopped = true; + if (timer) clearTimeout(timer); + // No outstanding beat is awaited here: a beat on a half-open connection would hold a finished + // upload behind its own budget for no decision the phase still has to make. + // A beat that ended the protection outranks a phase that settled meanwhile, from either side: the + // device is no longer ours, and the lease error is the reason the phase was not worth finishing. + if (terminalError !== undefined) throw terminalError; + if (!phase.ok) throw phase.error; + return phase.value; +} + +/** + * The inactivity window a beat just renewed, read from the lease its response carries. + * + * `heartbeatLease` answers with the lease, whose `expiresAt - heartbeatAt` is exactly the window it + * extended — the same pair `leaseOwnTtlMs` renews on. Anything unrecognizable leaves the caller on + * the fallback cadence rather than guessing one. + */ +function leaseWindowFromHeartbeatResponse(response: unknown): number | undefined { + const lease = ( + response as Readonly<{ data?: Readonly<{ lease?: Readonly> }> }> + )?.data?.lease; + const expiresAt = lease?.expiresAt; + const heartbeatAt = lease?.heartbeatAt; + if (typeof expiresAt !== 'number' || typeof heartbeatAt !== 'number') return undefined; + return expiresAt > heartbeatAt ? expiresAt - heartbeatAt : undefined; +} + +type Outcome = { ok: true; value: T } | { ok: false; error: unknown }; + +async function captureOutcome(promise: Promise): Promise> { + try { + return { ok: true, value: await promise }; + } catch (error) { + return { ok: false, error }; + } +} + +/** + * The beat that keeps a remote lease alive across a long client-side phase. + * + * `send` is the caller's transport and `budgetMs` is how long this beat may take: the beat's answer + * only matters before the next one is due, so a stalled round trip is cut off at the cadence rather + * than at the command's own 90-second heartbeat policy. A beat is only ever sent to a remote daemon, + * where a transport timeout performs no local cleanup. + */ +export function createLeaseRenewalBeat( + leaseScope: LeaseScope, + context: Readonly<{ + session: string; + sessionIsolation?: NonNullable['sessionIsolation']; + token: string; + send: (request: DaemonRequest, budgetMs: number) => Promise; + }>, +): (budgetMs: number) => Promise { + return async (budgetMs) => + await context.send( + buildLeaseHeartbeatRequest(leaseScope, { + session: context.session, + sessionIsolation: context.sessionIsolation, + requestId: createRequestId(), + token: context.token, + }), + budgetMs, + ); +} + +/** + * The request one beat sends: the command's lease scope, its own id, and nothing else. + * + * It is a fresh request rather than the install rewritten, so nothing about the upload — its source, + * its positional, its own request id — can be mistaken for the lease's state. The scope rides along + * as the command named it, so a beat renews for whatever window that command asked for, and for the + * lease's own window when it named none, which is the ordinary case for an install. Each beat gets a + * fresh id because a beat that times out is canceled under its own, and a shared id would let a later + * beat inherit an earlier cancellation. + */ +export function buildLeaseHeartbeatRequest( + leaseScope: LeaseScope, + context: Readonly<{ + session: string; + sessionIsolation?: NonNullable['sessionIsolation']; + requestId: string; + token: string; + }>, +): DaemonRequest { + return { + command: INTERNAL_COMMANDS.leaseHeartbeat, + positionals: [], + session: context.session, + token: context.token, + meta: { + ...leaseScopeToRequestMeta(leaseScope), + sessionIsolation: context.sessionIsolation, + requestId: context.requestId, + }, + }; +} + +/** The lease a request is running under, when it names one. */ +export function leaseScopeForHeartbeat( + request: Pick, +): LeaseScope | undefined { + const scope = leaseScopeFromRequest(request); + return scope.leaseId ? scope : undefined; +} + +/** + * The beat that renews a remote lease across an artifact upload, or `undefined` when there is no + * lease to protect: only a remote daemon uploads, so only one can be waiting on a billed device, and + * a command that names no lease has nothing to renew. + * + * Each beat's transport timeout is its budget, capped by the command's heartbeat policy: the answer + * is worthless once the next beat is due, so a stalled round trip is cut off and destroyed at the + * cadence instead of holding a socket for 90 seconds. `sendToDaemon` wires this once per upload. + */ +export function buildUploadLeaseHeartbeat( + info: DaemonInfo, + settings: DaemonClientSettings, + request: Omit, +): ((budgetMs: number) => Promise) | undefined { + if (!isRemoteDaemon(info)) return undefined; + const leaseScope = leaseScopeForHeartbeat(request); + if (!leaseScope) return undefined; + const policyTimeoutMs = resolveCommandRequestTimeoutMs( + resolveCommandTimeoutPolicy(INTERNAL_COMMANDS.leaseHeartbeat), + { positionals: [] }, + ); + return createLeaseRenewalBeat(leaseScope, { + session: request.session, + sessionIsolation: request.meta?.sessionIsolation, + token: info.token, + send: async (beat, budgetMs) => + await sendRequest( + info, + beat, + settings.transportPreference, + settings.paths, + // The beat's own budget governs; the command's heartbeat policy only ever caps it, and an + // unbounded policy leaves the budget standing on its own. + policyTimeoutMs === undefined ? budgetMs : Math.min(policyTimeoutMs, budgetMs), + ), + }); +} diff --git a/src/daemon-client/daemon-client.ts b/src/daemon-client/daemon-client.ts index 95a4ad4728..13610507eb 100644 --- a/src/daemon-client/daemon-client.ts +++ b/src/daemon-client/daemon-client.ts @@ -26,6 +26,7 @@ import { type EnsuredDaemon, } from './daemon-client-lifecycle.ts'; import { sendRequest } from './daemon-client-transport.ts'; +import { buildUploadLeaseHeartbeat, runProtectedLeaseWork } from './daemon-client-lease-beat.ts'; export type DaemonRequest = SharedDaemonRequest; export type DaemonResponse = SharedDaemonResponse; @@ -61,7 +62,10 @@ export async function sendToDaemon( { requestId, session: req.session }, ); const info = daemon.info; - const preparedRemoteRequest = await prepareRemoteRequestArtifacts(requestWithoutAuthFlag, info); + const preparedRemoteRequest = await runProtectedLeaseWork({ + heartbeat: buildUploadLeaseHeartbeat(info, settings, requestWithoutAuthFlag), + task: (signal) => prepareRemoteRequestArtifacts(requestWithoutAuthFlag, info, signal), + }); writeInstallInProgressNotice(requestWithoutAuthFlag.command); const request = buildTransportRequest( diff --git a/src/daemon/__tests__/lease-lifecycle.test.ts b/src/daemon/__tests__/lease-lifecycle.test.ts index e6c3f6751c..8fce339a62 100644 --- a/src/daemon/__tests__/lease-lifecycle.test.ts +++ b/src/daemon/__tests__/lease-lifecycle.test.ts @@ -48,7 +48,9 @@ test('admitRequestLeaseForLockedScope heartbeats and stores admitted lease on th expect(req.internal?.admittedLease?.leaseId).toBe(lease.leaseId); expect(req.internal?.admittedLease?.heartbeatAt).toBe(2_000); - expect(sessionStore.get('default')?.lease?.expiresAt).toBe(302_000); + // Renewal is for the window the lease carries — the registry default here, since this client + // named none — not the proxy-specific default admission used to name on every request (#2946). + expect(sessionStore.get('default')?.lease?.expiresAt).toBe(62_000); }); test('cleanupExpiredLeasedSession consumes expired lease and deletes the session after teardown', async () => { diff --git a/src/daemon/__tests__/lease-registry.test.ts b/src/daemon/__tests__/lease-registry.test.ts index fc3d576d5e..33852bfe44 100644 --- a/src/daemon/__tests__/lease-registry.test.ts +++ b/src/daemon/__tests__/lease-registry.test.ts @@ -646,6 +646,59 @@ test('releasing a lease drops the work claims recorded against it', () => { assert.equal(registry.listActiveLeases().length, 0, 'a released lease must not come back'); }); +// #2946: a caller that heartbeats without repeating its allocation TTL is asking for the same lease +// to keep going, not for the registry default. Resolving an absent `ttlMs` to that default silently +// shortened every lease allocated above it, which is how an upload expired the lease paying for the +// device it was uploading to. +test('a heartbeat with no ttlMs renews the lease for the window it already carries', () => { + let now = 1_000; + const registry = new LeaseRegistry({ now: () => now, defaultLeaseTtlMs: 10_000 }); + const lease = registry.allocateLease({ tenantId: 'tenant-a', runId: 'run-1', ttlMs: 60_000 }); + + now = 2_000; + const renewed = registry.heartbeatLease({ leaseId: lease.leaseId }); + assert.equal(renewed.heartbeatAt, 2_000); + assert.equal(renewed.expiresAt, 62_000, 'the allocated 60s window, not the 10s default'); +}); + +test('a heartbeat with no ttlMs keeps a long-TTL lease outliving the default TTL', () => { + let now = 0; + const registry = new LeaseRegistry({ now: () => now, defaultLeaseTtlMs: 60_000 }); + const lease = registry.allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'proxy', + deviceKey: 'android:mobile:emulator-5554', + ttlMs: 5 * 60_000, + }); + + // The 1m47s upload from the issue, beaten every 20s by the client and admitted after it lands. + for (const elapsed of [20_000, 40_000, 60_000, 80_000, 100_000, 107_000]) { + now = elapsed; + registry.heartbeatLease({ + leaseId: lease.leaseId, + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'proxy', + deviceKey: 'android:mobile:emulator-5554', + }); + } + + const active = registry.listActiveLeases().find((entry) => entry.leaseId === lease.leaseId); + assert.ok(active, 'the lease is still active at the end of the upload'); + assert.equal(active.expiresAt - active.heartbeatAt, 5 * 60_000); +}); + +test('a heartbeat with an explicit ttlMs still sets that window', () => { + let now = 1_000; + const registry = new LeaseRegistry({ now: () => now, defaultLeaseTtlMs: 10_000 }); + const lease = registry.allocateLease({ tenantId: 'tenant-a', runId: 'run-1', ttlMs: 60_000 }); + + now = 2_000; + const renewed = registry.heartbeatLease({ leaseId: lease.leaseId, ttlMs: 5_000 }); + assert.equal(renewed.expiresAt, 7_000, 'an explicit window is the caller asking to change it'); +}); + function inFlightClaimKeys(registry: LeaseRegistry): string[] { const work = ( registry as unknown as { diff --git a/src/daemon/__tests__/request-admission.test.ts b/src/daemon/__tests__/request-admission.test.ts index e4bfca61d0..c3c1a58e2e 100644 --- a/src/daemon/__tests__/request-admission.test.ts +++ b/src/daemon/__tests__/request-admission.test.ts @@ -164,6 +164,85 @@ test('sessionless apps admits a provider declared by the runtime app catalog', ( assert.equal(result, undefined); }); +// #2946 (the other half): a proxy client that allocated a lease with a window longer than the old +// synthetic admission default used to lose that window on the next admitted command — admission +// re-renewed every proxy lease at its own default instead of the window the lease carries. Admission +// now renews for the lease's window (ADR 0007); only a request that names a window changes it. +test('admitting a command does not shorten a proxy lease allocated above the old admission default', () => { + let now = 1_000; + const registry = new LeaseRegistry({ now: () => now }); + const lease = registry.allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'proxy', + deviceKey: 'ios:mobile:SIM-001', + clientId: 'client-a', + ttlMs: 600_000, + }); + now = 2_000; + + const result = assertRequestLeaseAdmission( + makeRequest({ + command: 'snapshot', + meta: { + tenantId: 'tenant-a', + runId: 'run-1', + sessionIsolation: 'tenant', + leaseId: lease.leaseId, + leaseProvider: 'proxy', + deviceKey: 'ios:mobile:SIM-001', + clientId: 'client-a', + }, + }), + registry, + undefined, + ); + + assert.equal(result?.leaseId, lease.leaseId); + // Without this, a lease handed back untouched would pass: allocation already leaves a 600_000 + // difference between these two stamps, so the delta alone cannot prove admission renewed anything. + assert.equal(result!.heartbeatAt, 2_000, 'admission renewed at admission time'); + assert.equal( + result!.expiresAt - result!.heartbeatAt, + 600_000, + 'the window the lease carries, not a smaller default', + ); +}); + +test('a command that names its own window renews the lease onto it', () => { + let now = 1_000; + const registry = new LeaseRegistry({ now: () => now }); + const lease = registry.allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'proxy', + deviceKey: 'ios:mobile:SIM-001', + clientId: 'client-a', + ttlMs: 600_000, + }); + now = 2_000; + + const result = assertRequestLeaseAdmission( + makeRequest({ + command: 'snapshot', + meta: { + tenantId: 'tenant-a', + runId: 'run-1', + sessionIsolation: 'tenant', + leaseId: lease.leaseId, + leaseProvider: 'proxy', + deviceKey: 'ios:mobile:SIM-001', + clientId: 'client-a', + leaseTtlMs: 90_000, + }, + }), + registry, + undefined, + ); + + assert.equal(result!.expiresAt - result!.heartbeatAt, 90_000); +}); + test('close still admits and heartbeats a real active lease', () => { let now = 1_000; const registry = new LeaseRegistry({ now: () => now }); diff --git a/src/daemon/__tests__/request-execution-scope.test.ts b/src/daemon/__tests__/request-execution-scope.test.ts index 4790083b8e..5da8da19d9 100644 --- a/src/daemon/__tests__/request-execution-scope.test.ts +++ b/src/daemon/__tests__/request-execution-scope.test.ts @@ -224,8 +224,8 @@ test('leased session admission uses stored lease metadata and heartbeats', async expect(scope.sessionName).toBe('default'); const activeLease = leaseRegistry.listActiveLeases()[0]; expect(activeLease?.heartbeatAt).toBe(2_000); - expect(activeLease?.expiresAt).toBe(302_000); - expect(sessionStore.get('default')?.lease?.expiresAt).toBe(302_000); + expect(activeLease?.expiresAt).toBe(62_000); + expect(sessionStore.get('default')?.lease?.expiresAt).toBe(62_000); }); test('leased session heartbeat is serialized with the request execution lock', async () => { diff --git a/src/daemon/__tests__/request-router-open.test.ts b/src/daemon/__tests__/request-router-open.test.ts index 344f7ba6b1..ec084f7c75 100644 --- a/src/daemon/__tests__/request-router-open.test.ts +++ b/src/daemon/__tests__/request-router-open.test.ts @@ -368,7 +368,7 @@ test('open stores admitted lease metadata on the session', async () => { leaseProvider: 'proxy', clientId: 'client-a', deviceKey: 'ios:SIM-LEASED', - expiresAt: 301_000, + expiresAt: 61_000, }); }); diff --git a/src/daemon/lease-context.ts b/src/daemon/lease-context.ts index 73d6c229f0..a07acb3d71 100644 --- a/src/daemon/lease-context.ts +++ b/src/daemon/lease-context.ts @@ -4,14 +4,13 @@ import type { DeviceLease } from '@agent-device/contracts/device'; import type { RunnerLogicalLeaseContext } from '@agent-device/contracts/runner-lease-context'; import { stripUndefined } from '@agent-device/kernel/record'; import { - DEFAULT_PROXY_LEASE_TTL_MS, findMissingProxyLeaseFields, isProxyLeaseScope, leaseScopeFromRequest, type LeaseScope, } from '@agent-device/contracts/lease-scope'; -export { DEFAULT_PROXY_LEASE_TTL_MS, findMissingProxyLeaseFields, isProxyLeaseScope }; +export { findMissingProxyLeaseFields, isProxyLeaseScope }; export type { LeaseScope }; export type SessionLease = { diff --git a/src/daemon/lease-registry-scope.ts b/src/daemon/lease-registry-scope.ts index 290bf86729..a228a9ec7f 100644 --- a/src/daemon/lease-registry-scope.ts +++ b/src/daemon/lease-registry-scope.ts @@ -328,6 +328,18 @@ export function createDeviceLease( }; } +/** + * The inactivity window a lease is currently living on: whatever TTL it was last renewed for. + * + * Every renewal that is not asked to change the window uses this, so a renewal never quietly + * re-decides it. Reading the registry default instead would shorten a lease allocated with a longer + * one — and a caller that heartbeats without repeating the allocation's TTL is not asking for the + * default, it is asking for the same lease to keep going. + */ +export function leaseOwnTtlMs(lease: Pick): number { + return lease.expiresAt - lease.heartbeatAt; +} + export function deviceLeaseBusyError(activeLease: DeviceLease): AppError { return new AppError('DEVICE_IN_USE', 'Device is already leased', { reason: 'DEVICE_LEASE_BUSY', diff --git a/src/daemon/lease-registry.ts b/src/daemon/lease-registry.ts index af8683d417..e0204ff6e1 100644 --- a/src/daemon/lease-registry.ts +++ b/src/daemon/lease-registry.ts @@ -23,6 +23,7 @@ import { assertLeaseOwnerScope, assertLeaseScopeMatch, leaseDeviceBindingKey, + leaseOwnTtlMs, leaseRunBindingKey, } from './lease-registry-scope.ts'; import { DeviceMutationDrain } from './device/device-mutation-drain.ts'; @@ -107,13 +108,21 @@ export class LeaseRegistry { return this.refreshLease(existingLease, leaseTtlMs); } + /** + * Extends a lease's life. A request naming a `ttlMs` asks for that inactivity window; one naming + * none renews for the window the lease already carries, the way protected work renews for its + * existing one (ADR 0007). Resolving an absent `ttlMs` to the registry default instead would + * shorten a lease every time a caller heartbeats without repeating its allocation TTL — which is + * what an admitted request does, and what expired the lease paying for a device mid-upload (#2946). + */ heartbeatLease(request: HeartbeatLeaseRequest): DeviceLease { const leaseId = normalizeRequiredLeaseId(request.leaseId); this.cleanupExpiredLeases(); const lease = this.getActiveLease(leaseId); assertLeaseOwnerScope(lease, request); assertLeaseScopeMatch(lease, request); - const leaseTtlMs = this.resolveLeaseTtlMs(request.ttlMs); + const leaseTtlMs = + request.ttlMs === undefined ? leaseOwnTtlMs(lease) : this.resolveLeaseTtlMs(request.ttlMs); return this.refreshLease(lease, leaseTtlMs); } @@ -427,11 +436,7 @@ export class LeaseRegistry { private refreshProtectedLease(leaseId: string | undefined, at: number): void { const lease = leaseId ? this.leases.get(leaseId) : undefined; if (lease) { - this.refreshLease( - lease, - lease.expiresAt - lease.heartbeatAt, - Math.max(at, lease.heartbeatAt), - ); + this.refreshLease(lease, leaseOwnTtlMs(lease), Math.max(at, lease.heartbeatAt)); } } diff --git a/src/daemon/request-admission.ts b/src/daemon/request-admission.ts index 12e8117b91..1d0bfd4490 100644 --- a/src/daemon/request-admission.ts +++ b/src/daemon/request-admission.ts @@ -8,9 +8,7 @@ import { } from './daemon-command-registry.ts'; import type { DeviceLease, ProviderAppCatalog } from '@agent-device/contracts/device'; import { - DEFAULT_PROXY_LEASE_TTL_MS, findMissingProxyLeaseFields, - isProxyLeaseScope, resolveLeaseScope, resolveRequestOrSessionLeaseScope, } from './lease-context.ts'; @@ -94,14 +92,11 @@ export function assertRequestLeaseAdmission( } assertRequestSessionLeaseMatches(requestLeaseScope, sessionLease); const leaseScope = resolveRequestOrSessionLeaseScope(req, session); - const heartbeatLeaseScope = { - ...leaseScope, - leaseTtlMs: - leaseScope.leaseTtlMs ?? - (isProxyLeaseScope(leaseScope) ? DEFAULT_PROXY_LEASE_TTL_MS : undefined), - }; leaseRegistry.assertLeaseAdmission(leaseScopeToHeartbeatRequest(leaseScope)); - const lease = leaseRegistry.heartbeatLease(leaseScopeToHeartbeatRequest(heartbeatLeaseScope)); + // Admission renews for the window the lease already carries, or the window this request named. + // Naming a proxy-specific default here used to shorten every lease allocated above it — a client + // that rented a device for longer than the default lost it on the next admitted command (#2946). + const lease = leaseRegistry.heartbeatLease(leaseScopeToHeartbeatRequest(leaseScope)); if (isHumanControlMutation(req)) leaseRegistry.assertHumanControlAdmission(lease); return lease; } diff --git a/src/remote/__tests__/daemon-artifacts-save-script.test.ts b/src/remote/__tests__/daemon-artifacts-save-script.test.ts index 9ecfd13c0e..137217d2ba 100644 --- a/src/remote/__tests__/daemon-artifacts-save-script.test.ts +++ b/src/remote/__tests__/daemon-artifacts-save-script.test.ts @@ -14,6 +14,7 @@ import { prepareRemoteRequestArtifacts } from '../daemon-artifacts.ts'; const REMOTE = { baseUrl: 'http://remote-mac.example.test:7777/agent-device', token: 'secret' }; const LOCAL = { token: 'secret' }; +const NO_CANCELLATION = new AbortController().signal; function replayRequest(saveScript?: boolean | string) { return { @@ -27,7 +28,7 @@ function replayRequest(saveScript?: boolean | string) { test('a remote daemon refuses --save-script, naming the caller/daemon split', async () => { await assert.rejects( - async () => await prepareRemoteRequestArtifacts(replayRequest(true), REMOTE), + async () => await prepareRemoteRequestArtifacts(replayRequest(true), REMOTE, NO_CANCELLATION), (error: unknown) => error instanceof AppError && error.code === 'INVALID_ARGS' && @@ -38,20 +39,24 @@ test('a remote daemon refuses --save-script, naming the caller/daemon split', as test('a remote daemon refuses an explicit --save-script output path too', async () => { await assert.rejects( async () => - await prepareRemoteRequestArtifacts(replayRequest('./flows/login.healed.ad'), REMOTE), + await prepareRemoteRequestArtifacts( + replayRequest('./flows/login.healed.ad'), + REMOTE, + NO_CANCELLATION, + ), (error: unknown) => error instanceof AppError && error.code === 'INVALID_ARGS', ); }); test('a local daemon leaves --save-script untouched', async () => { - const prepared = await prepareRemoteRequestArtifacts(replayRequest(true), LOCAL); + const prepared = await prepareRemoteRequestArtifacts(replayRequest(true), LOCAL, NO_CANCELLATION); assert.equal(prepared.flags?.saveScript, true); assert.deepEqual(prepared.positionals, ['./flows/login.ad']); }); test('a remote replay without --save-script passes through unchanged', async () => { - const prepared = await prepareRemoteRequestArtifacts(replayRequest(), REMOTE); + const prepared = await prepareRemoteRequestArtifacts(replayRequest(), REMOTE, NO_CANCELLATION); assert.deepEqual(prepared.positionals, ['./flows/login.ad']); assert.equal(prepared.uploadedArtifactId, undefined); diff --git a/src/remote/__tests__/daemon-artifacts-test-command.test.ts b/src/remote/__tests__/daemon-artifacts-test-command.test.ts index b804135b5c..898ecc3543 100644 --- a/src/remote/__tests__/daemon-artifacts-test-command.test.ts +++ b/src/remote/__tests__/daemon-artifacts-test-command.test.ts @@ -12,6 +12,7 @@ import { prepareRemoteRequestArtifacts } from '../daemon-artifacts.ts'; const REMOTE = { baseUrl: 'http://remote-mac.example.test:7777/agent-device', token: 'secret' }; const LOCAL = { token: 'secret' }; +const NO_CANCELLATION = new AbortController().signal; function testRequest(artifactsDir: string | undefined, cwd = '/repo') { return { @@ -27,6 +28,7 @@ test('a remote daemon redirects an explicit --artifacts-dir to a temp path it ow const prepared = await prepareRemoteRequestArtifacts( testRequest('remote-device-artifacts/ad-test'), REMOTE, + NO_CANCELLATION, ); const redirected = (prepared.flags as Record | undefined)?.artifactsDir; @@ -39,7 +41,11 @@ test('a remote daemon redirects an explicit --artifacts-dir to a temp path it ow }); test('a remote daemon redirects the default artifacts directory too', async () => { - const prepared = await prepareRemoteRequestArtifacts(testRequest(undefined), REMOTE); + const prepared = await prepareRemoteRequestArtifacts( + testRequest(undefined), + REMOTE, + NO_CANCELLATION, + ); const redirected = (prepared.flags as Record | undefined)?.artifactsDir; assert.equal(typeof redirected, 'string'); @@ -54,6 +60,7 @@ test('a remote daemon leaves an already-absolute --artifacts-dir as the download const prepared = await prepareRemoteRequestArtifacts( testRequest('/ci/artifacts/ad-test'), REMOTE, + NO_CANCELLATION, ); const redirected = (prepared.flags as Record | undefined)?.artifactsDir; @@ -65,6 +72,7 @@ test('a local daemon leaves --artifacts-dir untouched', async () => { const prepared = await prepareRemoteRequestArtifacts( testRequest('remote-device-artifacts/ad-test'), LOCAL, + NO_CANCELLATION, ); assert.equal( diff --git a/src/remote/daemon-artifacts.ts b/src/remote/daemon-artifacts.ts index c99a848f7b..4a56d088f7 100644 --- a/src/remote/daemon-artifacts.ts +++ b/src/remote/daemon-artifacts.ts @@ -31,6 +31,7 @@ type PreparedRemoteRequest = { export async function prepareRemoteRequestArtifacts( req: Omit, info: DaemonArtifactEndpoint, + signal: AbortSignal, ): Promise { const positionals = [...(req.positionals ?? [])]; let flags = req.flags ? { ...req.flags } : undefined; @@ -51,7 +52,7 @@ export async function prepareRemoteRequestArtifacts( assertRemoteDaemonSupportsSaveScript(req); flags = applyRemoteArtifactCommand(req, positionals, flags, clientArtifactPaths); - const remoteInstallSource = await prepareRemoteInstallSource(req, info, uploadProgress); + const remoteInstallSource = await prepareRemoteInstallSource(req, info, uploadProgress, signal); if (remoteInstallSource) { installSource = remoteInstallSource.installSource; uploadedArtifactId = remoteInstallSource.uploadedArtifactId ?? uploadedArtifactId; @@ -72,6 +73,7 @@ export async function prepareRemoteRequestArtifacts( info, positionals, uploadProgress, + signal, ); uploadedArtifactId = installPackageResult ?? uploadedArtifactId; return baseResult(); @@ -103,6 +105,7 @@ async function prepareRemoteInstallPackage( info: DaemonArtifactEndpoint, positionals: string[], onProgress: UploadProgressSink | undefined, + signal: AbortSignal, ): Promise { const pathIndex = positionals.length === 1 ? 0 : 1; const rawPath = positionals[pathIndex]; @@ -121,6 +124,7 @@ async function prepareRemoteInstallPackage( token: info.token, platform: req.flags?.platform, onProgress, + signal, }); } @@ -178,6 +182,7 @@ async function prepareRemoteInstallSource( req: Omit, info: DaemonArtifactEndpoint, onProgress: UploadProgressSink | undefined, + signal: AbortSignal, ): Promise<{ installSource: NonNullable['installSource']; uploadedArtifactId?: string; @@ -218,6 +223,7 @@ async function prepareRemoteInstallSource( token: info.token, platform: req.flags?.platform, onProgress, + signal, }); return { installSource: { diff --git a/src/remote/upload-client.ts b/src/remote/upload-client.ts index 25232764d6..c0a1be29b1 100644 --- a/src/remote/upload-client.ts +++ b/src/remote/upload-client.ts @@ -14,6 +14,11 @@ type UploadArtifactOptions = { token: string; platform?: string; onProgress?: UploadProgressSink; + /** + * Ends the upload when the thing it is uploading for stops being worth finishing — today, a beat + * that found the device's lease gone. Aborted requests reject with the signal's own reason. + */ + signal?: AbortSignal; }; type UploadResponse = { @@ -54,6 +59,7 @@ export async function uploadArtifact(options: UploadArtifactOptions): Promise; uploadAttemptId: string; onProgress?: UploadProgressSink; + signal?: AbortSignal; }): Promise { const uploadOnce = async ( preflight: Extract, ): Promise => { - await uploadDirectArtifact(options.artifact, preflight, options.onProgress); + await uploadDirectArtifact(options.artifact, preflight, options.onProgress, options.signal); return await finalizeDirectUpload({ normalizedBase: options.normalizedBase, token: options.token, uploadId: preflight.uploadId, + signal: options.signal, }); }; try { return await uploadOnce(options.preflight); } catch (error) { + // A canceled upload resumes for no reason: the thing it was uploaded for is already gone, and + // re-preflighting would ask the daemon for a fresh ticket to send bytes nobody is waiting on. + if (options.signal?.aborted) throw error; if (!shouldRetryDirectUpload(error)) return undefined; - const retryPreflight = await requestUploadPreflight({ - normalizedBase: options.normalizedBase, - token: options.token, - artifact: options.artifact, - uploadAttemptId: options.uploadAttemptId, - }); - if (retryPreflight?.kind === 'cache-hit') { - return retryPreflight.uploadId; - } - if (retryPreflight?.kind === 'direct-upload') { - try { - return await uploadOnce(retryPreflight); - } catch { - return undefined; - } + return await retryDirectUpload(options, uploadOnce); + } +} + +/** + * One fresh attempt after a retryable stream failure, asked for under the same attempt id. + * + * Anything short of a usable ticket — a cache hit excepted, which is itself a finished upload — + * hands the caller back to the legacy path. A second failure is not retried again: a ticket that + * failed twice is a transport or ticket problem the legacy route may still survive. + */ +async function retryDirectUpload( + options: { + normalizedBase: string; + token: string; + artifact: PreparedUploadArtifact; + uploadAttemptId: string; + onProgress?: UploadProgressSink; + signal?: AbortSignal; + }, + uploadOnce: ( + preflight: Extract, + ) => Promise, +): Promise { + const retryPreflight = await requestUploadPreflight({ + normalizedBase: options.normalizedBase, + token: options.token, + artifact: options.artifact, + uploadAttemptId: options.uploadAttemptId, + signal: options.signal, + }); + if (retryPreflight?.kind === 'cache-hit') { + return retryPreflight.uploadId; + } + if (retryPreflight?.kind === 'direct-upload') { + try { + return await uploadOnce(retryPreflight); + } catch { + return undefined; } - return undefined; } + return undefined; } function shouldRetryDirectUpload(error: unknown): boolean { @@ -147,6 +185,7 @@ async function uploadLegacyArtifact(options: { token: string; artifact: PreparedUploadArtifact; onProgress?: UploadProgressSink; + signal?: AbortSignal; }): Promise { const { normalizedBase, token, artifact } = options; const uploadUrl = new URL('upload', normalizedBase); @@ -170,6 +209,7 @@ async function uploadLegacyArtifact(options: { timeoutHint: 'The upload to the remote daemon exceeded the 5-minute timeout.', errorMessage: 'Failed to upload artifact to remote daemon', errorHint: 'Verify the remote daemon is reachable and supports artifact uploads.', + signal: options.signal, progress: { stage: 'legacy', fileName: artifact.fileName, @@ -194,6 +234,7 @@ async function requestUploadPreflight(options: { token: string; artifact: PreparedUploadArtifact; uploadAttemptId: string; + signal?: AbortSignal; }): Promise { const preflightUrl = new URL('upload/preflight', options.normalizedBase); const headers: Record = { @@ -204,7 +245,7 @@ async function requestUploadPreflight(options: { const response = await fetch(preflightUrl, { method: 'POST', headers, - signal: AbortSignal.timeout(UPLOAD_PREFLIGHT_TIMEOUT_MS), + signal: combineUploadSignals(options.signal, UPLOAD_PREFLIGHT_TIMEOUT_MS), body: JSON.stringify({ uploadAttemptId: options.uploadAttemptId, sha256: options.artifact.sha256, @@ -214,7 +255,12 @@ async function requestUploadPreflight(options: { ...(options.artifact.platform ? { platform: options.artifact.platform } : {}), contentType: options.artifact.contentType, }), - }).catch(() => undefined); + }).catch((error: unknown) => { + // A failed preflight is ordinary here — the caller falls back to the legacy upload. A canceled + // one is not: there is nothing to fall back to when the lease that paid for this upload is gone. + if (options.signal?.aborted) throw error; + return undefined; + }); if (!response?.ok) { return undefined; @@ -223,6 +269,15 @@ async function requestUploadPreflight(options: { return parseUploadPreflightResult(await response.json().catch(() => undefined)); } +/** + * The signal one upload request runs under: the caller's cancellation and the request's own + * timeout, whichever fires first. + */ +function combineUploadSignals(signal: AbortSignal | undefined, timeoutMs: number): AbortSignal { + const timeout = AbortSignal.timeout(timeoutMs); + return signal ? AbortSignal.any([signal, timeout]) : timeout; +} + function parseUploadPreflightResult(value: unknown): UploadPreflightResult | undefined { if (!value || typeof value !== 'object') { return undefined; @@ -265,6 +320,7 @@ async function uploadDirectArtifact( artifact: PreparedUploadArtifact, ticket: Extract, onProgress: UploadProgressSink | undefined, + signal: AbortSignal | undefined, ): Promise { const response = await streamFileToHttpRequest({ url: new URL(ticket.url), @@ -275,6 +331,7 @@ async function uploadDirectArtifact( timeoutHint: 'The direct upload ticket did not accept the artifact within the timeout.', errorMessage: 'Failed to upload artifact with direct upload ticket', retryable: true, + signal, progress: { stage: 'direct', fileName: artifact.fileName, @@ -294,6 +351,7 @@ async function finalizeDirectUpload(options: { normalizedBase: string; token: string; uploadId: string; + signal?: AbortSignal; }): Promise { const finalizeUrl = new URL('upload/finalize', options.normalizedBase); const headers: Record = { @@ -304,9 +362,12 @@ async function finalizeDirectUpload(options: { const response = await fetch(finalizeUrl, { method: 'POST', headers, - signal: AbortSignal.timeout(UPLOAD_PREFLIGHT_TIMEOUT_MS), + signal: combineUploadSignals(options.signal, UPLOAD_PREFLIGHT_TIMEOUT_MS), body: JSON.stringify({ uploadId: options.uploadId }), }).catch((error) => { + // As with preflight: a canceled finalize is not a transport failure to report, it is the caller + // saying this upload is no longer worth finishing. + if (options.signal?.aborted) throw error; throw new AppError('COMMAND_FAILED', 'Failed to finalize direct artifact upload', {}, error); }); diff --git a/src/remote/upload-stream.ts b/src/remote/upload-stream.ts index 608b48bdd8..fedb2c332d 100644 --- a/src/remote/upload-stream.ts +++ b/src/remote/upload-stream.ts @@ -36,6 +36,8 @@ export async function streamFileToHttpRequest(options: { errorMessage: string; errorHint?: string; retryable?: boolean; + /** Ends this request when the work the upload serves is over; see `uploadArtifact`. */ + signal?: AbortSignal; progress?: UploadStreamProgressOptions; }): Promise { return await streamFileToHttpRequestAttempt({ @@ -60,6 +62,7 @@ async function streamFileToHttpRequestAttempt(options: { errorMessage: string; errorHint?: string; retryable?: boolean; + signal?: AbortSignal; redirectCount: number; startOffset: number; progress?: UploadStreamProgressOptions; @@ -82,6 +85,9 @@ async function streamFileToHttpRequestAttempt(options: { method: options.method, path: options.url.pathname + options.url.search, headers, + // Aborting destroys the request mid-stream, which is the only way to stop bytes that are + // already piped at a device this client no longer holds the lease on. + ...(options.signal ? { signal: options.signal } : {}), }, (res) => { responseReceived = true; @@ -162,6 +168,12 @@ async function streamFileToHttpRequestAttempt(options: { req.on('error', (err) => { if (responseReceived) return; clearTimeout(timeout); + // A caller that canceled owns this outcome: the signal's reason is the answer, and wrapping it + // as an ordinary transport failure would let a retry path treat "we stopped" as "it broke". + if (options.signal?.aborted) { + reject(options.signal.reason); + return; + } reject( new AppError( 'COMMAND_FAILED', diff --git a/test/integration/provider-scenarios/remote-daemon-client.test.ts b/test/integration/provider-scenarios/remote-daemon-client.test.ts index 1b1ae2cb70..c80c5df38e 100644 --- a/test/integration/provider-scenarios/remote-daemon-client.test.ts +++ b/test/integration/provider-scenarios/remote-daemon-client.test.ts @@ -14,6 +14,8 @@ import { skipWhenLoopbackUnavailable, } from '../../../src/__tests__/test-utils/loopback.ts'; +const NO_CANCELLATION = new AbortController().signal; + type RemoteRpcRequest = { id: unknown; method?: string; @@ -678,6 +680,7 @@ test('remote web recording defaults client and daemon artifact paths to WebM', a meta: { cwd: '/tmp/project' }, }, { baseUrl: 'http://127.0.0.1:1', token: 'remote-token' }, + NO_CANCELLATION, ); assert.equal(prepared.positionals[0], 'start'); @@ -698,6 +701,7 @@ test('remote web recording appends WebM extension to extensionless client paths' meta: { cwd: '/tmp/project' }, }, { baseUrl: 'http://127.0.0.1:1', token: 'remote-token' }, + NO_CANCELLATION, ); assert.equal(prepared.positionals[0], 'start'); @@ -715,6 +719,7 @@ test('remote recording without platform or requested path lets daemon choose ses meta: { cwd: '/tmp/project' }, }, { baseUrl: 'http://127.0.0.1:1', token: 'remote-token' }, + NO_CANCELLATION, ); assert.deepEqual(prepared.positionals, ['start']); diff --git a/test/integration/provider-scenarios/remote-proxy-parity.test.ts b/test/integration/provider-scenarios/remote-proxy-parity.test.ts index 0cd8b69191..f0b99530c2 100644 --- a/test/integration/provider-scenarios/remote-proxy-parity.test.ts +++ b/test/integration/provider-scenarios/remote-proxy-parity.test.ts @@ -4,7 +4,6 @@ import http from 'node:http'; import os from 'node:os'; import path from 'node:path'; import { test } from 'vitest'; -import { DEFAULT_PROXY_LEASE_TTL_MS } from '@agent-device/contracts/lease-scope'; import { sendToDaemon } from '../../../src/daemon-client/daemon-client.ts'; import { LeaseRegistry } from '../../../src/daemon/lease-registry.ts'; import { createDaemonHttpServer } from '../../../src/daemon/server/http-server.ts'; @@ -549,7 +548,7 @@ test( async ({ world, proxied }) => { const session = 'leased'; const flags = { platform: 'ios', udid: SIM.id } as const; - const allocate = async (): Promise => { + const allocate = async (): Promise<{ leaseId: string; expiresAt: number }> => { const response = await proxied({ session, command: 'lease_allocate', @@ -558,9 +557,13 @@ test( meta: LEASE_SCOPE, }); assert.equal(response.ok, true, JSON.stringify(response)); - const leaseId = (response.ok ? response.data : {})?.lease as { leaseId?: string }; - assert.equal(typeof leaseId?.leaseId, 'string'); - return leaseId.leaseId!; + const lease = (response.ok ? response.data : {})?.lease as { + leaseId?: string; + expiresAt?: number; + }; + assert.equal(typeof lease?.leaseId, 'string'); + assert.equal(typeof lease?.expiresAt, 'number'); + return { leaseId: lease.leaseId!, expiresAt: lease.expiresAt! }; }; const run = async ( command: string, @@ -577,22 +580,24 @@ test( }); const firstLease = await allocate(); - assert.equal((await run('open', [APP], firstLease)).ok, true); + assert.equal((await run('open', [APP], firstLease.leaseId)).ok, true); assert.equal( - (await run('snapshot', [], firstLease, { snapshotInteractiveOnly: true })).ok, + (await run('snapshot', [], firstLease.leaseId, { snapshotInteractiveOnly: true })).ok, true, ); assert.equal( baselineInitialized( - await run('diff', ['snapshot'], firstLease, { snapshotInteractiveOnly: true }), + await run('diff', ['snapshot'], firstLease.leaseId, { snapshotInteractiveOnly: true }), ), false, 'the leased session holds comparison state before it expires', ); // The lease lapses without a heartbeat; the next request through the proxy finds it expired. - now += DEFAULT_PROXY_LEASE_TTL_MS + 1; - const expired = await run('diff', ['snapshot'], firstLease, { + // The jump is the window the lease itself was allocated with — a client that names no ttl + // gets the registry default, which is the only window the daemon promises it. + now = firstLease.expiresAt + 1; + const expired = await run('diff', ['snapshot'], firstLease.leaseId, { snapshotInteractiveOnly: true, }); assert.equal(expired.ok, false); @@ -603,16 +608,16 @@ test( assert.equal(world.daemon.session(session), undefined, 'expiry tears the session down'); const secondLease = await allocate(); - assert.notEqual(secondLease, firstLease); - assert.equal((await run('open', [APP], secondLease)).ok, true); + assert.notEqual(secondLease.leaseId, firstLease.leaseId); + assert.equal((await run('open', [APP], secondLease.leaseId)).ok, true); assert.equal( baselineInitialized( - await run('diff', ['snapshot'], secondLease, { snapshotInteractiveOnly: true }), + await run('diff', ['snapshot'], secondLease.leaseId, { snapshotInteractiveOnly: true }), ), true, 'a reacquired lease must not compare against the expired session tree', ); - assert.equal((await run('close', [], secondLease, {})).ok, true); + assert.equal((await run('close', [], secondLease.leaseId, {})).ok, true); }, { leaseRegistry }, ); diff --git a/test/integration/provider-scenarios/remote-upload-lease-beat.test.ts b/test/integration/provider-scenarios/remote-upload-lease-beat.test.ts new file mode 100644 index 0000000000..5c39cb45c5 --- /dev/null +++ b/test/integration/provider-scenarios/remote-upload-lease-beat.test.ts @@ -0,0 +1,351 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import http from 'node:http'; +import os from 'node:os'; +import path from 'node:path'; +import { test } from 'vitest'; +import { AppError } from '@agent-device/kernel/errors'; +import { sendToDaemon } from '../../../src/daemon-client/daemon-client.ts'; +import type { DaemonRequest } from '../../../src/daemon/daemon-request.ts'; +import { + listenOnLoopback, + skipWhenLoopbackUnavailable, +} from '../../../src/__tests__/test-utils/loopback.ts'; + +const LEASE_ID = 'lease-upload-beat'; +const TOKEN = 'upload-beat-token'; +const APK_BYTES = 4 * 1024 * 1024; +/** Renewed window the fake daemon reports; a third of it is the cadence the client beats on. */ +const LEASE_WINDOW_MS = 3_000; + +/** + * #2946's route end to end: `sendToDaemon` uploads an artifact for a remote install before the + * install request is admitted, and the beat is what keeps the lease alive across that gap. The unit + * suite covers the beat loop, the request a beat sends, and the upload client each in isolation; + * none of them would notice the wiring between the three being dropped. + */ + +type FakeDaemon = { + baseUrl: string; + /** Beats and commands the daemon saw, in arrival order. */ + seen: string[]; + uploadBytesDelivered(): number; + /** + * Lets a stalled artifact through, and reports how the upload ended: `drained` if the whole + * artifact arrived, `canceled` if the request carrying it was destroyed first. + * + * The caller releases it rather than a beat answering, so the answer is causal: the client has + * already been told the lease is gone by then, and an upload it did not cancel has nothing left + * that could stop it. + */ + releaseUploadAndObserveOutcome(graceMs?: number): Promise<'drained' | 'canceled' | 'unresolved'>; + close(): Promise; +}; + +type UploadBehaviour = 'complete' | 'backpressure'; + +/** + * A remote daemon that only answers a beat, and treats the upload as the long phase it is: + * + * - `complete` drains the artifact and withholds the upload response until a second beat has + * arrived, so "the lease was renewed while the artifact was still uploading" is a fact of the + * test rather than a race it happens to win. + * - `backpressure` stops reading after the first chunk, so the artifact is still in flight when the + * next beat reports the lease gone — the state the abort exists for. + */ +async function startFakeRemoteDaemon(behaviour: UploadBehaviour): Promise { + const seen: string[] = []; + let uploadBytesDelivered = 0; + let resolveOutcome!: (outcome: 'drained' | 'canceled') => void; + const settled = new Promise<'drained' | 'canceled'>((resolve) => { + resolveOutcome = resolve; + }); + let stopBackpressure: (() => void) | undefined; + let beatsAnswered = 0; + let leaseDeclaredLost = false; + let writeUploadResponse: (() => void) | undefined; + + const server = http.createServer((req, res) => { + if (req.method === 'GET' && (req.url ?? '').startsWith('/health')) { + writeJson(res, 200, { ok: true }); + return; + } + if (req.method !== 'POST') { + res.writeHead(404); + res.end('not found'); + return; + } + if (req.url === '/upload/preflight') { + // Draining before the 404 keeps the connection usable; an early end would make the client's + // fallback to the legacy upload route a matter of socket recycling rather than the protocol. + readJsonBody(req, () => { + res.writeHead(404); + res.end('not found'); + }); + return; + } + if (req.url === '/upload') { + handleUpload(req, res, behaviour, { + onBytes: (length) => { + uploadBytesDelivered += length; + }, + onBodyArrived: () => { + resolveOutcome('drained'); + }, + onCanceled: () => { + resolveOutcome('canceled'); + }, + holdResponse: (write) => { + writeUploadResponse = write; + if (!leaseDeclaredLost && beatsAnswered >= 2) write(); + }, + releaseBackpressure: (resume) => { + stopBackpressure = resume; + }, + }); + return; + } + if (req.url !== '/rpc') { + res.writeHead(404); + res.end('not found'); + return; + } + readJsonBody(req, (payload) => { + if (payload.method === 'agent_device.lease.heartbeat') { + answerBeat(res, payload); + return; + } + seen.push(String(payload.params?.command ?? payload.method)); + writeJson(res, 200, { + jsonrpc: '2.0', + id: payload.id, + result: { ok: true, data: { package: 'com.example.demo' } }, + }); + }); + }); + + function answerBeat(res: http.ServerResponse, payload: RpcPayload): void { + beatsAnswered += 1; + seen.push('lease_heartbeat'); + assert.equal(payload.params?.leaseId, LEASE_ID, 'a beat names the lease it protects'); + // Only a beat after the first can say anything about the upload: the loop fires one at t=0, + // while the artifact is still being hashed. + const leaseGone = behaviour === 'backpressure' && beatsAnswered >= 2; + if (leaseGone) { + leaseDeclaredLost = true; + writeLeaseLostError(res, payload.id); + // A beat that reports the lease gone must not also complete the upload it is meant to stop: + // the artifact's fate is decided by the abort, not by a response from here. + return; + } + const now = Date.now(); + writeJson(res, 200, { + jsonrpc: '2.0', + id: payload.id, + result: { ok: true, data: { lease: { heartbeatAt: now, expiresAt: now + LEASE_WINDOW_MS } } }, + }); + writeUploadResponse?.(); + } + server.keepAliveTimeout = 100; + + const port = await listenOnLoopback(server); + return { + baseUrl: `http://127.0.0.1:${String(port)}`, + seen, + uploadBytesDelivered: () => uploadBytesDelivered, + async releaseUploadAndObserveOutcome(graceMs = 2_000) { + stopBackpressure?.(); + return await Promise.race([ + settled, + new Promise<'unresolved'>((resolve) => { + setTimeout(() => resolve('unresolved'), graceMs).unref(); + }), + ]); + }, + async close() { + server.closeAllConnections(); + await new Promise((resolve) => { + server.close(() => resolve()); + }); + }, + }; +} + +function handleUpload( + req: http.IncomingMessage, + res: http.ServerResponse, + behaviour: UploadBehaviour, + hooks: Readonly<{ + onBytes(length: number): void; + onBodyArrived(): void; + onCanceled(): void; + holdResponse(write: () => void): void; + releaseBackpressure(resume: () => void): void; + }>, +): void { + let answered = false; + const respond = (): void => { + if (answered) return; + answered = true; + writeJson(res, 200, { ok: true, uploadId: 'upload-demo.apk' }); + }; + const requestSettled = (): void => { + if (answered) return; + hooks.onCanceled(); + }; + req.on('aborted', requestSettled); + res.on('close', requestSettled); + let stalled = false; + req.on('data', (chunk: Buffer) => { + hooks.onBytes(chunk.length); + // Stopping the read applies backpressure, so the artifact stays in flight instead of racing + // through loopback and making "stopped early" a matter of timing. + // Pausing once, rather than on every chunk: releasing the pressure has to actually let the + // artifact through, otherwise a stalled upload and a canceled one are the same observation. + if (behaviour === 'backpressure' && !stalled) { + req.pause(); + stalled = true; + } + }); + req.on('end', () => { + hooks.onBodyArrived(); + hooks.holdResponse(respond); + }); + hooks.releaseBackpressure(() => { + if (stalled) req.resume(); + }); +} + +type RpcPayload = Readonly<{ id: unknown; method: string; params?: Record }>; + +function readJsonBody(req: http.IncomingMessage, done: (payload: RpcPayload) => void): void { + let body = ''; + req.setEncoding('utf8'); + req.on('data', (chunk) => { + body += chunk; + }); + req.on('end', () => { + done(JSON.parse(body) as RpcPayload); + }); +} + +function writeJson(res: http.ServerResponse, statusCode: number, payload: unknown): void { + res.writeHead(statusCode, { 'content-type': 'application/json' }); + res.end(JSON.stringify(payload)); +} + +function writeLeaseLostError(res: http.ServerResponse, id: unknown): void { + writeJson(res, 400, { + jsonrpc: '2.0', + id, + error: { + code: -32000, + message: 'Lease is not active', + data: { + code: 'UNAUTHORIZED', + message: 'Lease is not active', + details: { reason: 'LEASE_NOT_FOUND' }, + }, + }, + }); +} + +function installRequest( + baseUrl: string, + apkPath: string, + stateDir: string, +): Omit { + return { + session: 'upload-beat', + command: 'install', + positionals: [apkPath], + flags: { + platform: 'android', + daemonBaseUrl: baseUrl, + stateDir, + leaseId: LEASE_ID, + tenant: 'acme', + runId: 'run-1', + leaseProvider: 'proxy', + deviceKey: 'android:mobile:emulator-5554', + clientId: 'client-a', + }, + meta: { cwd: path.dirname(apkPath) }, + }; +} + +const TRANSPORT = { authToken: TOKEN } as const; + +async function withUploadFixture( + t: { skip(reason?: string): void }, + behaviour: UploadBehaviour, + run: (daemon: FakeDaemon, apkPath: string) => Promise, +): Promise { + if (await skipWhenLoopbackUnavailable(t)) return undefined; + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'agent-device-upload-beat-')); + const apkPath = path.join(dir, 'demo.apk'); + fs.writeFileSync(apkPath, Buffer.alloc(APK_BYTES, 'x')); + const daemon = await startFakeRemoteDaemon(behaviour); + try { + return await run(daemon, apkPath); + } finally { + await daemon.close(); + fs.rmSync(dir, { recursive: true, force: true }); + } +} + +test('an install beats the lease while its artifact uploads, before the install RPC', async (t) => { + await withUploadFixture(t, 'complete', async (daemon, apkPath) => { + const response = await sendToDaemon( + installRequest(daemon.baseUrl, apkPath, path.dirname(apkPath)), + TRANSPORT, + ); + + assert.equal(response.ok, true); + assert.ok( + daemon.seen.filter((entry) => entry === 'lease_heartbeat').length >= 2, + `a beat has to land while the upload is held open, daemon saw: ${daemon.seen.join(', ')}`, + ); + assert.equal( + daemon.uploadBytesDelivered(), + APK_BYTES, + 'the artifact arrived whole on a lease that was being renewed under it', + ); + assert.deepEqual(daemon.seen, ['lease_heartbeat', 'lease_heartbeat', 'install']); + }); +}); + +test('a lease lost mid-upload aborts the upload and no install request goes out', async (t) => { + await withUploadFixture(t, 'backpressure', async (daemon, apkPath) => { + await assert.rejects( + async () => + await sendToDaemon( + installRequest(daemon.baseUrl, apkPath, path.dirname(apkPath)), + TRANSPORT, + ), + (error: unknown) => + error instanceof AppError && + error.code === 'UNAUTHORIZED' && + error.details?.reason === 'LEASE_NOT_FOUND', + ); + + assert.ok( + daemon.uploadBytesDelivered() > 0 && daemon.uploadBytesDelivered() < APK_BYTES, + `bytes were in flight and the daemon stopped reading at ${String( + daemon.uploadBytesDelivered(), + )} of ${String(APK_BYTES)}, which is what had to be stopped`, + ); + // The daemon never read past the first chunk, so this is the only way the artifact can stop + // short: the client destroyed the request. An upload nobody canceled drains once the pressure + // comes off, and by now the client has long since been told the lease is gone. + assert.equal( + await daemon.releaseUploadAndObserveOutcome(), + 'canceled', + 'the upload was stopped, not left to finish on a lease nobody held', + ); + assert.ok( + !daemon.seen.includes('install'), + `nothing was asked of a device no longer ours, daemon saw: ${daemon.seen.join(', ')}`, + ); + }); +}); diff --git a/test/wire-compat/ledger.json b/test/wire-compat/ledger.json index 181e04ba82..b61d956aaf 100644 --- a/test/wire-compat/ledger.json +++ b/test/wire-compat/ledger.json @@ -161,14 +161,14 @@ "src/remote/upload-client.ts#UploadPreflightResponse": "sha256:196a1d9ef41a1aec192589fa7ccb30cf71d971a005349d5d73eed42e6ab21327", "src/remote/upload-client.ts#UploadPreflightResult": "sha256:bd8fc6d617c0630325455fd64d0fe319daaced198ee6745e4673a8ace3df460d", "src/remote/upload-client.ts#UploadResponse": "sha256:499e01466b5e0e03413f8d3f1986f5af6bc120704c50a3747d84de422b3f0314", - "src/remote/upload-client.ts#finalizeDirectUpload": "sha256:5e32c87214f9f6140535b48d05de8565cb121370fdc334c05f7f0ad476b73a07", + "src/remote/upload-client.ts#finalizeDirectUpload": "sha256:5d283353df5f089a6ab26b7be96c44997e6b9905b329cb993d300c615479d715", "src/remote/upload-client.ts#isStringRecord": "sha256:b6d6b571b90baf0976d9c347c4a45a4363a4df5ad184ea1087cb3b9febd4b624", "src/remote/upload-client.ts#parseUploadPreflightResult": "sha256:7ce3f43b45c03df8b7f2690432beedc7a537c1bba540961716862f8a632d5623", - "src/remote/upload-client.ts#requestUploadPreflight": "sha256:df42c0eaef3a54f4388c99d74bf6030885239780c2ddc78117151dbd391afedd", + "src/remote/upload-client.ts#requestUploadPreflight": "sha256:4c9863bce8757f1dc0f01ec7374019e752efc8e786cd070200f516a1d72d09e1", "src/remote/upload-client.ts#shouldRetryDirectUpload": "sha256:29059f076ff7f2b9540ad15408be48792b2d63a601ef2b9fb2d54f5576d49bbd", - "src/remote/upload-client.ts#tryDirectUploadWithResume": "sha256:0f57cf9bb1eefa3b988e5d723b2eebf2c4a622c3f7c808f35eae070771acafa6", - "src/remote/upload-client.ts#uploadDirectArtifact": "sha256:5f26f8f7b6ada2fe2089a214fb8350e3a9f65725212952c17f6787b3ca8fe232", - "src/remote/upload-client.ts#uploadLegacyArtifact": "sha256:f3ccab9c5ba18fc2f2000ea8e2a4bbb3cc97f7b50f6111192e513013f635cdd7", + "src/remote/upload-client.ts#tryDirectUploadWithResume": "sha256:67596a2ff2b8a5402566c3b7b99a6fa1baaf9d5de3d647bf35398185cc45eea6", + "src/remote/upload-client.ts#uploadDirectArtifact": "sha256:c16cb5dd74895931b8e41e16629a0519e5601340563c976b7f800c6de8d158b2", + "src/remote/upload-client.ts#uploadLegacyArtifact": "sha256:a7a82d6eb2925d6f510b9cc00f01f9ce865638368e0bec503b700f0bc59871bc", "src/remote/upload-stream.ts#MAX_UPLOAD_REDIRECTS": "sha256:12ff46dfe33a1dce95bdb79c90c17a3ba4f3919e7dc4b3a1b04590fff430f79f", "src/remote/upload-stream.ts#UploadStreamResponse": "sha256:724e284c2a6bd7b4b3a49b1d8c61c8923d6cc24777c3d76224a69d190e56adce", "src/remote/upload-stream.ts#buildUploadRequestHeaders": "sha256:827576b13a9f3e6cea445119a6a29b2ac5eaa23b66436cf810b2d09765c82a4c", @@ -177,8 +177,8 @@ "src/remote/upload-stream.ts#isUploadResumeStatus": "sha256:9f730ccd1c3401c326f8df67d1fd8243fef757357a38a1f53736df2b90bd8fdc", "src/remote/upload-stream.ts#parseNonNegativeIntegerHeader": "sha256:5cbe9c6994459d673d988a51407e65556e28175b187ce0a9d27cbb6331962834", "src/remote/upload-stream.ts#parseUploadResumeOffset": "sha256:297184ea22a6dc1cda9aee17034122ae0ea2f43d0355647f6fcf02f3f4b5ca0e", - "src/remote/upload-stream.ts#streamFileToHttpRequest": "sha256:eaf2ea49957034f6a7e17d34092dc6cb1998812bddfd68381a0ac34d8a2de58c", - "src/remote/upload-stream.ts#streamFileToHttpRequestAttempt": "sha256:da39a79fa7c1f81e55caf613eedc347c0f3a9a9711b265a4db185677532d9552" + "src/remote/upload-stream.ts#streamFileToHttpRequest": "sha256:ce07ea33a275e06e4cceaf0c75a079bd0c7938f36df6817407dbb1cc8c4ff900", + "src/remote/upload-stream.ts#streamFileToHttpRequestAttempt": "sha256:aa73fb51890ca5e43d4aa80c1d2124b574568380a73b19eae7da0ee3e3e7acf8" }, "compatibleChanges": [ { @@ -300,6 +300,41 @@ "declaration": "packages/kernel/src/contracts.ts#daemonRuntimeSchema", "digest": "sha256:3b99926ce9deb5d66186a55b1168011dc1ae4d010ada5ac5d79c4c7225a1f80f", "rationale": "#2266 broadens runtime-hint validation to accept the additive HarmonyOS platform value; existing protocol-2 runtime hints remain valid and unchanged." + }, + { + "declaration": "src/remote/upload-client.ts#requestUploadPreflight", + "digest": "sha256:4c9863bce8757f1dc0f01ec7374019e752efc8e786cd070200f516a1d72d09e1", + "rationale": "#2946 runs the preflight fetch under the caller's abort signal in addition to its own timeout. Route, headers, and JSON body are unchanged, so a protocol-2 daemon parses an uncancelled preflight exactly as before; an aborted one is the client disconnect it already handles." + }, + { + "declaration": "src/remote/upload-client.ts#uploadDirectArtifact", + "digest": "sha256:c16cb5dd74895931b8e41e16629a0519e5601340563c976b7f800c6de8d158b2", + "rationale": "#2946 threads an optional abort signal into the PUT so a lost lease can stop the stream. Method, ticket url and headers, and body bytes are unchanged for an uncancelled upload; an abort truncates the request, which a protocol-2 daemon sees as the client disconnect it already handles." + }, + { + "declaration": "src/remote/upload-client.ts#tryDirectUploadWithResume", + "digest": "sha256:67596a2ff2b8a5402566c3b7b99a6fa1baaf9d5de3d647bf35398185cc45eea6", + "rationale": "#2946 stops a canceled upload from re-preflighting and lifts the retry leg into its own function. The request sequence a daemon sees — preflight, direct PUT, finalize, or the legacy fallback after a retryable failure — is unchanged." + }, + { + "declaration": "src/remote/upload-client.ts#finalizeDirectUpload", + "digest": "sha256:5d283353df5f089a6ab26b7be96c44997e6b9905b329cb993d300c615479d715", + "rationale": "#2946 adds the caller's abort signal to finalize's existing timeout and lets a canceled one reject with the signal's own reason instead of a wrapped transport error. Route and body keys are unchanged, so a protocol-2 daemon parses an uncancelled finalize exactly as before." + }, + { + "declaration": "src/remote/upload-client.ts#uploadLegacyArtifact", + "digest": "sha256:a7a82d6eb2925d6f510b9cc00f01f9ce865638368e0bec503b700f0bc59871bc", + "rationale": "#2946 threads an optional abort signal into the legacy upload stream. Method, route, artifact headers, and chunked body are unchanged for an uncancelled upload; an abort truncates the request, which a protocol-2 daemon sees as the client disconnect it already handles." + }, + { + "declaration": "src/remote/upload-stream.ts#streamFileToHttpRequest", + "digest": "sha256:ce07ea33a275e06e4cceaf0c75a079bd0c7938f36df6817407dbb1cc8c4ff900", + "rationale": "#2946 widens the options with an optional abort signal handed to node:http.request. The request line, headers built by buildUploadRequestHeaders, and piped payload are unchanged, so an older daemon's 200/308 handling is unaffected." + }, + { + "declaration": "src/remote/upload-stream.ts#streamFileToHttpRequestAttempt", + "digest": "sha256:aa73fb51890ca5e43d4aa80c1d2124b574568380a73b19eae7da0ee3e3e7acf8", + "rationale": "#2946 passes the optional abort signal to the request and rejects with that signal's reason when it is what ended the stream. The released 308 resume contract is untouched: offset headers, resume start offsets, redirects, and payload bytes are computed exactly as before." } ] } diff --git a/test/wire-compat/module-resolution.ts b/test/wire-compat/module-resolution.ts index f281315690..605ebaba0f 100644 --- a/test/wire-compat/module-resolution.ts +++ b/test/wire-compat/module-resolution.ts @@ -35,6 +35,7 @@ export type ResolvedOrigin = * `DaemonRequest`), so treating the constructor as a leaf loses nothing. */ const TS_GLOBALS = new Set([ + 'AbortSignal', 'Array', 'Awaited', 'Date', diff --git a/website/docs/docs/remote-proxy.md b/website/docs/docs/remote-proxy.md index 593a2a8ae0..1668f1716c 100644 --- a/website/docs/docs/remote-proxy.md +++ b/website/docs/docs/remote-proxy.md @@ -45,7 +45,7 @@ agent-device disconnect Passing `--daemon-auth-token ` instead of exporting the environment variable also works, but only authenticates the single command it is passed to; subsequent commands need the token again through the env var, a `daemonAuthToken` entry in your remote config profile, or a repeated `--daemon-auth-token` flag. -`connect proxy` stores the proxy profile and client identity. Device leases are automatic on `open` and expire after five minutes without commands. `close` releases the active session and device lease; `disconnect` clears local connection state. +`connect proxy` stores the proxy profile and client identity. Device leases are automatic on `open` and expire after five minutes without commands. That five minutes is the window `open` asks for; a lease allocated directly over the RPC without `ttlMs` keeps the daemon's one-minute inactivity default instead. `close` releases the active session and device lease; `disconnect` clears local connection state. Multiple agents can share one proxy when each uses the normal `connect proxy`, `open`, commands, `close`, and `disconnect` flow. A busy device error means another agent owns the device until it closes or its inactivity lease expires.