diff --git a/docs/adr/0007-remote-device-leases.md b/docs/adr/0007-remote-device-leases.md index d46af73fe0..4302946f89 100644 --- a/docs/adr/0007-remote-device-leases.md +++ b/docs/adr/0007-remote-device-leases.md @@ -91,7 +91,10 @@ A daemon that starts and finds this journal hands each lease in it to the expire It reads both lease records only after it holds the daemon registration for the state directory, because until then they can belong to a daemon that is still running. The lease itself is not restored, and its controller connects again. Restoring the live lease was rejected: the sessions and device claims of the dead daemon are gone, so a restored lease would name a device that no session can reach. -A lease that leaves the registry for expiry stays in the journal until its expired-lease record is durable, and a journal write that fails is retried. +A lease that leaves the registry for expiry stays in the journal until its expired-lease record is durable, and a journal write that fails is retried until this daemon shuts down. +If a journal retry is pending, daemon shutdown makes one final journal write. It does not schedule another journal retry. +A daemon that has shut down writes lease records only synchronously. It arms no timer for them, because a timer could fire after that daemon released its registration, when the records belong to the next daemon. +After the daemon shuts its lease releaser down, that releaser refuses every further provider-backed release. It still writes the expired-lease record of a recoverable lease, so the next daemon inherits it. If no record names the lease, the daemon logs the refusal with its lease provider, because no daemon will end that device. Two routes still let the registry forget a provider lease while its device may be rented. The journal and expiry recovery both follow the registry, so neither covers it, and only the provider runtime still holding the device can end it: - A failed first allocation releases the registry entry without a provider release, while its provider start may still run. diff --git a/packages/daemon/src/server/__tests__/provider-lease-expiry.test.ts b/packages/daemon/src/server/__tests__/provider-lease-expiry.test.ts index 79f54d7176..543221e5b6 100644 --- a/packages/daemon/src/server/__tests__/provider-lease-expiry.test.ts +++ b/packages/daemon/src/server/__tests__/provider-lease-expiry.test.ts @@ -150,10 +150,11 @@ test('shutdown drain reports releases completed before and during its final retr }); try { - await releaser.release(releasedBeforeShutdown); + await expect(releaser.release(releasedBeforeShutdown)).resolves.toEqual({ status: 'released' }); releaser.beginShutdown(); - await releaser.release(releasedDuringShutdown); - await releaser.release(lease); + // This one's first attempt fails, so the releaser keeps it and the drain retries it. + await expect(releaser.release(releasedDuringShutdown)).resolves.toEqual({ status: 'retrying' }); + await expect(releaser.release(lease)).resolves.toEqual({ status: 'released' }); const drained = await releaser.drain(10); expect(release).toHaveBeenCalledTimes(4); @@ -389,6 +390,144 @@ test('daemon startup recovers inherited lease records only after it holds the re expect(source.indexOf('expiredProviderLeaseReleaser.retryPending()')).toBe(-1); }); +test('a release that fails after shutdown is not retried and stays durable', async () => { + vi.useFakeTimers(); + const stateDir = mkdtempForTestSync('agent-device-expired-provider-lease-'); + const lease = new LeaseRegistry().allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'limrun', + }); + let rejectHeldRelease: ((error: Error) => void) | undefined; + const recoverExpiredLease = vi.fn( + () => + new Promise((_resolve, reject) => { + rejectHeldRelease = reject; + }), + ); + const releaser = createExpiredProviderLeaseReleaser({ + recoverExpiredLease, + stateDir, + recoverableProviderIds: ['limrun'], + retryDelayMs: 1_000, + }); + + try { + releaser.beginShutdown(); + // The provider call is still in flight when the drain budget runs out, so its outcome is unknown. + const admitted = releaser.release(lease); + const drained = releaser.drain(10); + await vi.advanceTimersByTimeAsync(10); + expect(await drained).toEqual({ pending: [lease], unknown: [lease], released: [] }); + expect(recoverExpiredLease).toHaveBeenCalledTimes(1); + + // Plugins are disposed here. The provider call this daemon is waiting on then fails. + releaser.shutdown(); + rejectHeldRelease!(new Error('provider plugin disposed')); + await vi.advanceTimersByTimeAsync(60_000); + + expect(recoverExpiredLease).toHaveBeenCalledTimes(1); + // Nothing is armed to call the provider again, and nothing is armed to write a lease record. + expect(vi.getTimerCount()).toBe(0); + const record = JSON.parse( + fs.readFileSync(`${stateDir}/expired-provider-leases.json`, 'utf8'), + ) as { leases: DeviceLease[] }; + expect(record.leases.map((entry) => entry.leaseId)).toEqual([lease.leaseId]); + // The provider call this daemon was waiting on failed once the releaser had closed, so the lease + // stayed pending, and nothing here will attempt it again. The record is what carries it forward. + await expect(admitted).resolves.toEqual({ status: 'refused', reason: 'closed', durable: true }); + } finally { + releaser.shutdown(); + vi.useRealTimers(); + fs.rmSync(stateDir, { recursive: true, force: true }); + } +}); + +test('release after shutdown starts no provider call and retries nothing', async () => { + vi.useFakeTimers(); + const stateDir = mkdtempForTestSync('agent-device-expired-provider-lease-'); + const liveOnlyLease = new LeaseRegistry().allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'browserstack', + }); + const recoverableLease = new LeaseRegistry().allocateLease({ + tenantId: 'tenant-a', + runId: 'run-2', + leaseProvider: 'limrun', + }); + const release = vi.fn(async (_lease: DeviceLease) => ({})); + const recoverExpiredLease = vi.fn(async (_lease: DeviceLease) => undefined); + const releaser = createExpiredProviderLeaseReleaser({ + leaseLifecycleProvider: { release }, + providerRuntimeIds: ['browserstack'], + recoverExpiredLease, + stateDir, + recoverableProviderIds: ['limrun'], + retryDelayMs: 10, + }); + + try { + releaser.shutdown(); + await expect(releaser.release(liveOnlyLease)).resolves.toEqual({ + status: 'refused', + reason: 'closed', + durable: false, + }); + await expect(releaser.release(recoverableLease)).resolves.toEqual({ + status: 'refused', + reason: 'closed', + durable: true, + }); + await vi.advanceTimersByTimeAsync(60_000); + + expect(release).not.toHaveBeenCalled(); + expect(recoverExpiredLease).not.toHaveBeenCalled(); + // What a closed daemon cannot release, it leaves durable for the next one. + const record = JSON.parse( + fs.readFileSync(`${stateDir}/expired-provider-leases.json`, 'utf8'), + ) as { leases: DeviceLease[] }; + expect(record.leases.map((entry) => entry.leaseId)).toEqual([recoverableLease.leaseId]); + } finally { + releaser.shutdown(); + vi.useRealTimers(); + fs.rmSync(stateDir, { recursive: true, force: true }); + } +}); + +test('a lease that expires after shutdown is still recovered by the next daemon', async () => { + const stateDir = mkdtempForTestSync('agent-device-live-provider-lease-'); + let now = 1_000; + const recoverExpiredLease = vi.fn(async (_lease: DeviceLease) => undefined); + const options = { recoverExpiredLease, stateDir, recoverableProviderIds: ['revyl'] }; + const daemon = createExpiredProviderLeaseReleaser(options); + const registry = new LeaseRegistry({ + defaultLeaseTtlMs: 10, + minLeaseTtlMs: 1, + now: () => now, + onLeaseExpired: (lease) => void daemon.release(lease), + onProviderLeasesChanged: (leases) => daemon.recordLiveLeases(leases), + }); + + try { + const lease = registry.allocateLease({ tenantId: 't', runId: 'run-1', leaseProvider: 'revyl' }); + daemon.shutdown(); + // An abandoned request let this lease expire once this daemon had stopped releasing. + now = 1_011; + registry.consumeExpiredLease(lease.leaseId); + + expect(recoverExpiredLease).not.toHaveBeenCalled(); + + const restartedDaemon = createExpiredProviderLeaseReleaser(options); + await restartedDaemon.recoverInherited(); + expect(recoverExpiredLease).toHaveBeenCalledWith(lease); + restartedDaemon.shutdown(); + } finally { + daemon.shutdown(); + fs.rmSync(stateDir, { recursive: true, force: true }); + } +}); + test('an unreadable live journal is reported with its path and left on disk', async () => { const stateDir = mkdtempForTestSync('agent-device-live-provider-lease-'); const journalPath = `${stateDir}/live-provider-leases.json`; diff --git a/packages/daemon/src/server/http/daemon-lease-finalizer.test.ts b/packages/daemon/src/server/http/daemon-lease-finalizer.test.ts index 0d56c8ec56..e07ce82bd1 100644 --- a/packages/daemon/src/server/http/daemon-lease-finalizer.test.ts +++ b/packages/daemon/src/server/http/daemon-lease-finalizer.test.ts @@ -3,6 +3,7 @@ import path from 'node:path'; import { expect, test, vi } from 'vitest'; import { LeaseRegistry } from '../../resources/lease-registry.ts'; import type { DeviceLease } from '@agent-device/contracts/device'; +import { withDaemonDiagnosticsScope } from '@agent-device/daemon-contracts/daemon-diagnostics-scope'; import { createExpiredProviderLeaseReleaser } from '../provider-lease-expiry.ts'; import { finalizeDaemonLeases } from './daemon-lease-finalizer.ts'; import { mkdtempForTestSync } from '@agent-device/test-kit'; @@ -52,6 +53,62 @@ test('journals and bounds a hung recoverable lease release before the final drai } }); +test('a release the closed releaser refused is reported and not counted as completed', async () => { + const stateDir = mkdtempForTestSync('agent-device-daemon-lease-finalizer-'); + const leaseRegistry = new LeaseRegistry(); + const lease = leaseRegistry.allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'browserstack', + }); + const release = vi.fn(async (_lease: DeviceLease) => ({})); + const expiredProviderLeaseReleaser = createExpiredProviderLeaseReleaser({ + leaseLifecycleProvider: { release }, + providerRuntimeIds: ['browserstack'], + }); + // The drain budget ran out and the releaser closed before this lease was offered to it. + expiredProviderLeaseReleaser.beginShutdown(); + expiredProviderLeaseReleaser.shutdown(); + + const logPath = path.join(stateDir, 'daemon.log'); + try { + const outcomes = await withDaemonDiagnosticsScope({ logPath }, () => + finalizeDaemonLeases({ + leaseRegistry, + expiredProviderLeaseReleaser, + timeoutMs: 1_000, + }), + ); + + expect(release).not.toHaveBeenCalled(); + // The daemon stops either way, so the registry entry goes; the summary and the log say this daemon + // never handed the device to its provider. + expect(outcomes.refused).toEqual([lease.leaseId]); + expect(leaseRegistry.listActiveLeases()).toEqual([]); + const rows = fs + .readFileSync(logPath, 'utf8') + .trim() + .split('\n') + .map((line) => JSON.parse(line) as { phase: string; data?: Record }); + expect(rows).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + level: 'error', + phase: 'provider_lease_release_refused', + data: { + leaseId: lease.leaseId, + reason: 'closed', + durable: false, + leaseProvider: 'browserstack', + }, + }), + ]), + ); + } finally { + fs.rmSync(stateDir, { recursive: true, force: true }); + } +}); + test('final drain joins a release that completes after the lease timeout', async () => { vi.useFakeTimers(); const leaseRegistry = new LeaseRegistry(); diff --git a/packages/daemon/src/server/http/daemon-lease-finalizer.ts b/packages/daemon/src/server/http/daemon-lease-finalizer.ts index 2fe9cb0c47..7ae7384d1a 100644 --- a/packages/daemon/src/server/http/daemon-lease-finalizer.ts +++ b/packages/daemon/src/server/http/daemon-lease-finalizer.ts @@ -1,55 +1,99 @@ import { emitDiagnostic } from '@agent-device/host-kit/diagnostics'; -import type { ExpiredProviderLeaseReleaser } from '../provider-lease-expiry.ts'; +import type { DeviceLease } from '@agent-device/contracts/device'; +import { + leaseReleaseDiagnosticEvent, + type ExpiredProviderLeaseReleaser, + type ProviderLeaseReleaseOutcome, +} from '../provider-lease-expiry.ts'; import type { LeaseRegistry } from '../../resources/lease-registry.ts'; import { leaseReleaseRequestFor } from '../../resources/lease-registry-scope.ts'; +/** + * The leases this finalizer was refused, by id. The shutdown summary reports these: a release its + * provider never saw is not a release this daemon completed. + */ +export type DaemonLeaseReleaseSummary = Readonly<{ + refused: readonly string[]; +}>; + +/** What one lease came to here. The lease travels with it: the report must name it whichever way this went. */ +type LeaseReleaseSettlement = + | { lease: DeviceLease; outcome: ProviderLeaseReleaseOutcome } + | { lease: DeviceLease; timedOut: true } + | { lease: DeviceLease; error: unknown }; + /** * Ends every lease still active at daemon shutdown, whether a session held it or its owner kept it * through `close` with `retainOnClose`. Each provider release is bounded by `timeoutMs`; one that - * runs past it stays with the releaser for the shutdown drain. + * runs past it stays with the releaser for the shutdown drain. A release the closed releaser refused is + * reported per lease and named in the returned summary, so the shutdown says the device was left with + * the provider instead of reading as a completed release. */ export async function finalizeDaemonLeases(params: { leaseRegistry: LeaseRegistry; expiredProviderLeaseReleaser: ExpiredProviderLeaseReleaser; timeoutMs: number; -}): Promise { +}): Promise { const { leaseRegistry, expiredProviderLeaseReleaser, timeoutMs } = params; - await Promise.all( - leaseRegistry.listActiveLeases().map(async (lease) => { + const settlements = await Promise.all( + leaseRegistry.listActiveLeases().map(async (lease): Promise => { try { - const completed = await releaseWithinTimeout( + const outcome = await releaseWithinTimeout( expiredProviderLeaseReleaser.release(lease), timeoutMs, ); + // The daemon stops either way, so the registry entry goes with it. What carries a lease past this + // process is its recovery record and its diagnostic, never this registry. leaseRegistry.releaseLease(leaseReleaseRequestFor(lease)); - if (!completed) { - emitDiagnostic({ - level: 'warn', - phase: 'daemon_shutdown_lease_release_timed_out', - data: { leaseId: lease.leaseId, timeoutMs }, - }); - } + if (outcome === undefined) return { lease, timedOut: true }; + return { lease, outcome }; } catch (error) { - emitDiagnostic({ - level: 'warn', - phase: 'daemon_shutdown_lease_release_failed', - data: { - leaseId: lease.leaseId, - error: error instanceof Error ? error.message : String(error), - }, - }); + return { lease, error }; } }), ); + const refused: string[] = []; + for (const settlement of settlements) { + const { leaseId } = settlement.lease; + // A release whose budget ran out is reported here and answered by the drain, which owns the attempt. + if ('timedOut' in settlement) { + emitDiagnostic({ + level: 'warn', + phase: 'daemon_shutdown_lease_release_timed_out', + data: { leaseId, timeoutMs }, + }); + continue; + } + if ('error' in settlement) { + emitDiagnostic({ + level: 'warn', + phase: 'daemon_shutdown_lease_release_failed', + data: { + leaseId, + error: + settlement.error instanceof Error ? settlement.error.message : String(settlement.error), + }, + }); + continue; + } + // The mapping owns the event, so a refusal reads the same here as it does on the expiry route. + const event = leaseReleaseDiagnosticEvent(settlement.lease, settlement.outcome); + if (event) emitDiagnostic(event); + if (settlement.outcome.status === 'refused') refused.push(leaseId); + } + return Object.freeze({ refused: Object.freeze(refused) }); } -async function releaseWithinTimeout(release: Promise, timeoutMs: number): Promise { - return await new Promise((resolve, reject) => { - const timeout = setTimeout(() => resolve(false), timeoutMs); +async function releaseWithinTimeout( + release: Promise, + timeoutMs: number, +): Promise { + return await new Promise((resolve, reject) => { + const timeout = setTimeout(() => resolve(undefined), timeoutMs); void release.then( - () => { + (value) => { clearTimeout(timeout); - resolve(true); + resolve(value); }, (error: unknown) => { clearTimeout(timeout); diff --git a/packages/daemon/src/server/http/daemon-runtime-lease-expiry.test.ts b/packages/daemon/src/server/http/daemon-runtime-lease-expiry.test.ts index e059b1d2f8..5f224b66cb 100644 --- a/packages/daemon/src/server/http/daemon-runtime-lease-expiry.test.ts +++ b/packages/daemon/src/server/http/daemon-runtime-lease-expiry.test.ts @@ -1,6 +1,7 @@ import assert from 'node:assert/strict'; import fs from 'node:fs'; import { afterEach, test, vi } from 'vitest'; +import type { DeviceLease } from '@agent-device/contracts/device'; const leaseProbe = vi.hoisted(() => ({ registries: [] as import('../../resources/lease-registry.ts').LeaseRegistry[], @@ -84,6 +85,74 @@ test('an expired lease ends its session without any further request', async () = } }); +test('a lease that expires after shutdown is refused and reported, not sent to its provider', async () => { + vi.useFakeTimers(); + const stateDir = mkdtempForTestSync('agent-device-daemon-lease-closed-'); + const release = vi.fn(async (_lease: DeviceLease) => ({})); + try { + const runtime = await startDaemonRuntime({ + composeDeps: daemonRuntimeDepsFixture({ + leaseLifecycleProvider: { release }, + providerRuntimeIds: ['browserstack'], + }), + host: createTestHostEnvironment({ + env: { + ...process.env, + AGENT_DEVICE_STATE_DIR: stateDir, + AGENT_DEVICE_DAEMON_SERVER_MODE: 'http', + AGENT_DEVICE_DAEMON_IDLE_TIMEOUT_MS: '0', + AGENT_DEVICE_LEASE_MIN_TTL_MS: '1', + }, + }), + stderr: { write: () => {} }, + stdout: { write: () => {} }, + }); + assert.notEqual(runtime, null); + const [registry] = leaseProbe.registries; + assert.ok(registry); + const closing = runtime!.close(); + await vi.runOnlyPendingTimersAsync(); + await closing; + + // An abandoned request let this lease go past its TTL once this daemon had stopped releasing. + const lease = registry.allocateLease({ + tenantId: 'tenant-a', + runId: 'run-1', + leaseProvider: 'browserstack', + ttlMs: 10, + }); + await vi.advanceTimersByTimeAsync(1_000); + registry.consumeExpiredLease(lease.leaseId); + await vi.advanceTimersByTimeAsync(0); + + // The provider was never asked, and the daemon log names the lease it could not end. + assert.deepEqual(release.mock.calls, []); + const rows = fs + .readFileSync(`${stateDir}/daemon.log`, 'utf8') + .split('\n') + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as { phase: string; level: string; data?: unknown }); + assert.ok( + rows.some( + (row) => + row.phase === 'provider_lease_release_refused' && + row.level === 'error' && + JSON.stringify(row.data) === + JSON.stringify({ + leaseId: lease.leaseId, + reason: 'closed', + durable: false, + leaseProvider: 'browserstack', + }), + ), + `expected a refusal row in the daemon log, got ${JSON.stringify(rows.map((r) => r.phase))}`, + ); + } finally { + vi.useRealTimers(); + fs.rmSync(stateDir, { recursive: true, force: true }); + } +}); + test('shutdown cancels and drains lease ends before it tears sessions down', async () => { const closeAdmission = SessionStore.prototype.closeAdmission; vi.spyOn(SessionStore.prototype, 'closeAdmission').mockImplementation( diff --git a/packages/daemon/src/server/http/daemon-runtime.ts b/packages/daemon/src/server/http/daemon-runtime.ts index 17b8409138..a0f72ca015 100644 --- a/packages/daemon/src/server/http/daemon-runtime.ts +++ b/packages/daemon/src/server/http/daemon-runtime.ts @@ -29,7 +29,10 @@ import { type InteractorResolution, } from '../../core/interactor-resolution.ts'; import { LeaseRegistry } from '../../resources/lease-registry.ts'; -import { createExpiredProviderLeaseReleaser } from '../provider-lease-expiry.ts'; +import { + createExpiredProviderLeaseReleaser, + leaseReleaseDiagnosticEvent, +} from '../provider-lease-expiry.ts'; import { createRequestHandler, type RequestRouterDeps } from '../request-router.ts'; import { createLeaseSessionEnd, @@ -379,7 +382,24 @@ async function startDaemonRuntimeOnHost( minLeaseTtlMs: parseIntegerEnv(env.AGENT_DEVICE_LEASE_MIN_TTL_MS), maxLeaseTtlMs: parseIntegerEnv(env.AGENT_DEVICE_LEASE_MAX_TTL_MS), onLeaseExpired: (lease) => { - void expiredProviderLeaseReleaser.release(lease); + void expiredProviderLeaseReleaser.release(lease).then( + (outcome) => { + // A lease can expire after shutdown closed the releaser, and the sweep runs outside any + // request, so this report goes to the daemon log rather than a request scope. + const event = leaseReleaseDiagnosticEvent(lease, outcome); + if (event) { + void emitDaemonDiagnostic(logPath, event.phase, event.data, event.level).catch( + () => undefined, + ); + } + }, + (error: unknown) => { + void emitDaemonDiagnostic(logPath, 'provider_lease_release_outcome_failed', { + leaseId: lease.leaseId, + error: error instanceof Error ? error.message : String(error), + }).catch(() => undefined); + }, + ); void endSessionsForLease(lease.leaseId, 'LEASE_EXPIRED'); }, onProviderLeasesChanged: (leases) => expiredProviderLeaseReleaser.recordLiveLeases(leases), @@ -840,13 +860,13 @@ async function startDaemonRuntimeOnHost( } catch {} expiredProviderLeaseReleaser.beginShutdown(); const sessions = await teardownDaemonSessions(); - await withDaemonDiagnosticsScope({ logPath }, async () => { - await finalizeDaemonLeases({ + const leaseFinalization = await withDaemonDiagnosticsScope({ logPath }, async () => + finalizeDaemonLeases({ leaseRegistry, expiredProviderLeaseReleaser, timeoutMs: DAEMON_LEASE_RELEASE_TIMEOUT_MS, - }); - }); + }), + ); try { await deps.ownerLifecycle.resetAndroidSnapshotHelper(); } catch (error) { @@ -865,6 +885,9 @@ async function startDaemonRuntimeOnHost( data: { releasedLeaseIds: providerReleaseDrain.released.map((lease) => lease.leaseId), pendingLeaseIds: providerReleaseDrain.pending.map((lease) => lease.leaseId), + // The drain only sees work it still holds, so the refusals the finalizer was given are reported + // here as well: they are leases this daemon will not end, and this is the shutdown's one summary. + refusedLeaseIds: [...leaseFinalization.refused], releasedDeviceKeys: shutdownClaimLedger.claims.released.map((claim) => claim.deviceKey), orphanedDeviceKeys: shutdownClaimLedger.claims.orphaned.map((claim) => claim.deviceKey), }, diff --git a/packages/daemon/src/server/http/daemon-startup-diagnostics.ts b/packages/daemon/src/server/http/daemon-startup-diagnostics.ts index b47c2a73cc..a91a44d103 100644 --- a/packages/daemon/src/server/http/daemon-startup-diagnostics.ts +++ b/packages/daemon/src/server/http/daemon-startup-diagnostics.ts @@ -1,5 +1,9 @@ import { normalizeError } from '@agent-device/kernel/errors'; -import { emitDiagnostic, type ResourceDiagnostic } from '@agent-device/host-kit/diagnostics'; +import { + emitDiagnostic, + type DiagnosticLevel, + type ResourceDiagnostic, +} from '@agent-device/host-kit/diagnostics'; import type { OwnerIdentity } from '@agent-device/host-kit/process'; import type { DaemonRegistrationOwner } from '@agent-device/daemon-contracts/daemon-registration-owner'; import { withDaemonDiagnosticsScope } from '@agent-device/daemon-contracts/daemon-diagnostics-scope'; @@ -21,14 +25,20 @@ export async function flushDaemonStartupDiagnostics( }); } -/** Records one daemon-level event outside any request scope, with debug forced on. */ +/** + * Records one daemon-level event outside any request scope, with debug forced on. + * `level` is the event's own severity; it defaults to `warn` because these events are all daemon + * shutdown or recovery problems, and a caller passes another only to keep a severity its own owner + * already decided. + */ export async function emitDaemonDiagnostic( logPath: string, phase: string, data: Record, + level: DiagnosticLevel = 'warn', ): Promise { await withDaemonDiagnosticsScope({ logPath }, () => { - emitDiagnostic({ level: 'warn', phase, data }); + emitDiagnostic({ level, phase, data }); }); } diff --git a/packages/daemon/src/server/provider-lease-expiry.ts b/packages/daemon/src/server/provider-lease-expiry.ts index 41f85e331f..4e66859e22 100644 --- a/packages/daemon/src/server/provider-lease-expiry.ts +++ b/packages/daemon/src/server/provider-lease-expiry.ts @@ -30,14 +30,50 @@ const PERSISTED_LEASE_NUMBER_FIELDS = ['createdAt', 'heartbeatAt', 'expiresAt'] type Timer = ReturnType; +/** + * The lifecycle of one releaser. `running` admits releases and re-arms retries. `draining` is the + * bounded shutdown window in which the final release attempts run. `closed` starts no provider attempt + * and arms no timer: its providers may already be disposed, and a timer could fire after this daemon + * released its registration and rewrite records the next daemon owns. It still writes lease records + * synchronously, because they are how the next daemon learns what to recover. + */ +type ReleaserPhase = 'running' | 'draining' | 'closed'; + +const PHASE_RANK: Record = { running: 0, draining: 1, closed: 2 }; + +/** Why a release was refused. `closed` names the shutdown boundary; `provider-runtime-unavailable` names a daemon that never had the runtime to end the lease. */ +type ReleaseRefusalReason = 'closed' | 'provider-runtime-unavailable'; + +/** + * What one `release` call did with the lease it was given. + * + * `released` means this daemon ended the lease with its provider. `retrying` means this daemon still + * owns the lease and will attempt it again. `not-required` means no provider ever held it. `refused` + * means this daemon will not end this lease: `reason` says why, and `durable` says whether a lease + * record now names it for the next daemon. A refused lease whose `durable` is false is a device no + * daemon will end, so every caller reports it. + */ +export type ProviderLeaseReleaseOutcome = + | { status: 'released' } + | { status: 'retrying' } + | { status: 'not-required' } + | { status: 'refused'; reason: ReleaseRefusalReason; durable: boolean }; + type PersistedProviderLeaseRecord = { version: number; leases: DeviceLease[]; }; export type ExpiredProviderLeaseReleaser = { + /** Opens the bounded shutdown window: the final release attempts, and the accounting that feeds the drain report. */ beginShutdown: () => void; - release: (lease: DeviceLease) => Promise; + /** + * Ends one lease. Admits it while this releaser runs or is draining, and answers with what it did. + * Once closed it refuses: no provider attempt starts and no pending work is admitted, because the + * providers may already be disposed. A recoverable lease is still made durable for the next daemon, + * which is what the outcome's `durable` answers. + */ + release: (lease: DeviceLease) => Promise; /** Journals the registry's live provider leases, so a daemon killed while holding them recovers them. */ recordLiveLeases: (leases: readonly DeviceLease[]) => void; /** @@ -76,11 +112,19 @@ export function createExpiredProviderLeaseReleaser(options: { const retryDelayMs = options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS; let retryTimer: Timer | undefined; const activeReleaseAttempts = new Map>(); - let trackingShutdownReleases = false; - const shutdownReleasedLeases = new Map(); + let phase: ReleaserPhase = 'running'; + /** The releases this drain window observed; replaced at each window so one window's set is never another's. */ + let drainReleasedLeases = new Map(); + + /** Moves the lifecycle forward and nothing else: the phase never goes back. */ + const advancePhase = (next: ReleaserPhase): void => { + if (PHASE_RANK[next] > PHASE_RANK[phase]) phase = next; + }; const recordReleasedLease = (lease: DeviceLease): void => { - if (trackingShutdownReleases) shutdownReleasedLeases.set(lease.leaseId, lease); + // The drain window is the only period that reports releases, so the phase is the tracker and the + // set cannot grow outside it. + if (phase === 'draining') drainReleasedLeases.set(lease.leaseId, lease); }; const persistPendingLeases = (): boolean => { @@ -156,6 +200,9 @@ export function createExpiredProviderLeaseReleaser(options: { journalRetryTimer = undefined; } catch (error) { emitJournalFailure('provider_lease_journal_write_failed', journal.size, error); + // The retry timer is not armed once closed: it could fire after this daemon released its + // registration, and writes made then belong to the daemon that holds it. + if (phase === 'closed') return; journalRetryTimer ??= setTimeout(() => { journalRetryTimer = undefined; writeJournal(); @@ -189,6 +236,9 @@ export function createExpiredProviderLeaseReleaser(options: { const scheduleRetry = (): void => { if ( + // A closed releaser retries nothing: its providers may already be disposed, and the next daemon + // inherits whatever these lease records still hold. + phase === 'closed' || retryTimer || (!hasRetryableLiveLease(pendingLiveLeases, options.providerRuntimeIds) && !hasRetryableRecoveryLease(pendingRecoveryLeases, options.recoverableProviderIds)) @@ -227,6 +277,7 @@ export function createExpiredProviderLeaseReleaser(options: { }; const ensureLiveReleaseAttempt = (lease: DeviceLease): Promise | undefined => { + if (phase === 'closed') return undefined; if (!isProviderRuntimeAvailable(lease, options.providerRuntimeIds)) return undefined; const key = `live:${lease.leaseId}`; const activeAttempt = activeReleaseAttempts.get(key); @@ -241,6 +292,7 @@ export function createExpiredProviderLeaseReleaser(options: { }; const ensureRecoveryReleaseAttempt = (lease: DeviceLease): Promise | undefined => { + if (phase === 'closed') return undefined; if (!isRecoverableProviderAvailable(lease, options.recoverableProviderIds)) return undefined; if (!persistedRecoveryLeaseIds.has(lease.leaseId) && !persistPendingLeases()) return undefined; const key = `recovery:${lease.leaseId}`; @@ -268,13 +320,45 @@ export function createExpiredProviderLeaseReleaser(options: { await Promise.all(attempts); }; + /** + * Refuses a lease that arrives once this releaser is closed. Nothing is awaited and no attempt is + * tracked, because that work belongs to the daemon that now owns the state directory. A recoverable + * lease is still written out: `pendingRecoveryLeases` is the content of its lease record, so writing + * through the map is what makes the handoff durable, and the outcome says so. + */ + const refuseClosedRelease = (lease: DeviceLease): ProviderLeaseReleaseOutcome => { + if (!isRecoverableProviderAvailable(lease, options.recoverableProviderIds)) { + return { status: 'refused', reason: 'closed', durable: false }; + } + pendingRecoveryLeases.set(lease.leaseId, lease); + return { status: 'refused', reason: 'closed', durable: persistPendingLeases() }; + }; + + /** What the attempt this release started came to. The pending entry is the provider's answer: it is removed only when the provider confirmed the end. */ + const releaseOutcomeFor = ( + lease: DeviceLease, + pendingLeases: ReadonlyMap, + ): ProviderLeaseReleaseOutcome => { + if (!pendingLeases.has(lease.leaseId)) return { status: 'released' }; + // The releaser closed while the provider call was in flight, so nothing here will attempt it again. + if (phase === 'closed') { + return { + status: 'refused', + reason: 'closed', + durable: persistedRecoveryLeaseIds.has(lease.leaseId), + }; + } + return { status: 'retrying' }; + }; + return { beginShutdown: () => { - shutdownReleasedLeases.clear(); - trackingShutdownReleases = true; + advancePhase('draining'); + drainReleasedLeases = new Map(); }, release: async (lease) => { - if (!lease.leaseProvider) return; + if (!lease.leaseProvider) return { status: 'not-required' }; + if (phase === 'closed') return refuseClosedRelease(lease); if (isRecoverableProviderAvailable(lease, options.recoverableProviderIds)) { // Set before the first await: the registry journals right after `onLeaseExpired` returns, and the // journal keeps this lease only while it is pending here and not yet durable. @@ -284,15 +368,19 @@ export function createExpiredProviderLeaseReleaser(options: { } else { scheduleRetry(); } - return; + return releaseOutcomeFor(lease, pendingRecoveryLeases); + } + if (!isProviderRuntimeAvailable(lease, options.providerRuntimeIds)) { + return { status: 'refused', reason: 'provider-runtime-unavailable', durable: false }; + } + if (!options.leaseLifecycleProvider?.release) { + return { status: 'refused', reason: 'provider-runtime-unavailable', durable: false }; } - if (!isProviderRuntimeAvailable(lease, options.providerRuntimeIds)) return; - if (!options.leaseLifecycleProvider?.release) return; pendingLiveLeases.set(lease.leaseId, lease); await ensureLiveReleaseAttempt(lease); - if (pendingLiveLeases.has(lease.leaseId)) { - scheduleRetry(); - } + const outcome = releaseOutcomeFor(lease, pendingLiveLeases); + if (outcome.status === 'retrying') scheduleRetry(); + return outcome; }, recordLiveLeases, recoverInherited, @@ -311,24 +399,53 @@ export function createExpiredProviderLeaseReleaser(options: { activeReleaseAttempts.has(`recovery:${lease.leaseId}`), ), ], - released: [...shutdownReleasedLeases.values()], + released: [...drainReleasedLeases.values()], }; - trackingShutdownReleases = false; - shutdownReleasedLeases.clear(); + drainReleasedLeases = new Map(); return result; }, shutdown: () => { + advancePhase('closed'); if (retryTimer) clearTimeout(retryTimer); retryTimer = undefined; if (journalRetryTimer) { // One last attempt, so a journal that a retry was still owed does not outlive this daemon stale. + // The closed phase keeps this the last lease record write that is owed a retry. clearTimeout(journalRetryTimer); journalRetryTimer = undefined; writeJournal(); } - // A last attempt that failed re-armed the timer; nothing runs it after shutdown. - if (journalRetryTimer) clearTimeout(journalRetryTimer); - journalRetryTimer = undefined; + }, + }; +} + +/** One diagnostic event a caller reports, as `emitDiagnostic` takes it. */ +export type LeaseReleaseDiagnosticEvent = { + level: 'warn' | 'error'; + phase: string; + data: Record; +}; + +/** + * The diagnostic a refusal needs, or `undefined` for any other outcome. Every caller of `release` + * reports through this mapping, so a refusal cannot look different depending on which path noticed it. + * A refusal whose lease is not durable names a device no daemon will end, so it is an error and carries + * the lease provider: that is where an operator ends the device by hand. A release the releaser still + * owns is already reported by the retry that owns it, and a completed one by its provider. + */ +export function leaseReleaseDiagnosticEvent( + lease: DeviceLease, + outcome: ProviderLeaseReleaseOutcome, +): LeaseReleaseDiagnosticEvent | undefined { + if (outcome.status !== 'refused') return undefined; + return { + level: outcome.durable ? 'warn' : 'error', + phase: 'provider_lease_release_refused', + data: { + leaseId: lease.leaseId, + reason: outcome.reason, + durable: outcome.durable, + ...(lease.leaseProvider ? { leaseProvider: lease.leaseProvider } : {}), }, }; } diff --git a/packages/host-kit/src/diagnostics.ts b/packages/host-kit/src/diagnostics.ts index 3f69c72b4d..841dd3d421 100644 --- a/packages/host-kit/src/diagnostics.ts +++ b/packages/host-kit/src/diagnostics.ts @@ -3,6 +3,7 @@ export { createRequestId, DIAGNOSTIC_STDERR_PREFIX, type DiagnosticEventInput, + type DiagnosticLevel, emitDiagnostic, flushDiagnosticsToSessionFile, getDiagnosticsMeta, diff --git a/packages/host-kit/src/internal/diagnostics.ts b/packages/host-kit/src/internal/diagnostics.ts index 9855bec828..1095fb2250 100644 --- a/packages/host-kit/src/internal/diagnostics.ts +++ b/packages/host-kit/src/internal/diagnostics.ts @@ -6,7 +6,8 @@ import path from 'node:path'; import { redactDiagnosticData } from '@agent-device/kernel/redaction'; import type { DiagnosticsRecordRef } from '@agent-device/kernel/errors'; -type DiagnosticLevel = 'info' | 'warn' | 'error' | 'debug'; +/** The severity an event is recorded with. */ +export type DiagnosticLevel = 'info' | 'warn' | 'error' | 'debug'; type DiagnosticEvent = { ts: string;