diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 39dc659d..8fd0e9f4 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -31,6 +31,7 @@ edit yourself keeps the mode you give it until the next `config set`; run | `lease.requestRetentionMs` | How long a settled lease request stays stored, counted from when it settled. A client that repeats a request with the same `idempotencyKey` inside this window gets the stored result back; after it, the record is removed and the same key starts a new request. | `10 minutes` | | `lease.maxRequestRecords` | The most lease-request records Simlock keeps. When a new request would go over it, the oldest settled record is removed. A request that is still open is never removed. | `10000` | | `lease.maxPerRequester` | How many leases one agent id may hold at once. A positive integer; anything else (`0`, `-1`, `1.5`, a string) stops the daemon from starting. An agent id still has at most one waiting request at a time. At the limit, or with a request already waiting, a new request is refused with `REQUESTER_ALREADY_LEASED`. On a gateway the limit counts the agent's leases across the whole fleet, and workers do not apply their own limit to what the gateway sends them. An agent that holds more than the limit (you lowered it and restarted) keeps its leases and cannot request more until it is below. Read at start; changing it needs a restart. **Risk above `1`:** agents that lease one device after another can block each other, each holding one device while waiting for a second.| `1` | +| `lease.endWaitMs` | How long Simlock waits, after a lease ends, for add-ons that act on the device before it is wiped (for example one that ends a session on it). A positive number of milliseconds, at most `2147483647`; anything else stops the daemon from starting. When it passes, the wipe starts anyway. Stopping the daemon and `simlock nuke` do not wait. With no such add-on, nothing waits. Read at start; changing it needs a restart. | `600000` | | `gateway.url` | **Worker side.** Base URL of the gateway this worker joins, e.g. `wss://gw.example:4700`; the worker dials `/v1/uplink` and upgrades it to a WebSocket. Joining grants that gateway the `admin` role on this daemon, and `gateway.token` rides on the upgrade request as a bearer credential, so use `wss://` — or plain `ws://` only over loopback or inside your own tunnel. `http://` and `https://` are rejected at load. Unset means "do not join a fleet": the default, and the only thing that changes about a joined worker. | unset | | `gateway.token` | **Worker side.** The join token (`simlock token create --role worker`, minted on the gateway) this worker presents when it opens its uplink. Required whenever `gateway.url` is set. `simlock config` shows it; the daemon log and the `daemon.started` event show `[redacted]`. It is stored in plain text in `config.json`, which `simlock config set` writes owner-only. | unset | | `gateway.label` | **Worker side.** Display name for this worker in `simlock worker list`, `status`, the console, and on the lease's `worker` block. Display-only: nothing routes on it and it need not be unique. | the worker's own id | diff --git a/docs/internal/COMPONENTS.md b/docs/internal/COMPONENTS.md index c2e41b90..6a239f38 100644 --- a/docs/internal/COMPONENTS.md +++ b/docs/internal/COMPONENTS.md @@ -101,7 +101,7 @@ modules with. | `AcquisitionPlanner` | `src/leasing/acquisition-planner.ts` | The read-only plan for one request (for a physical request: the first fitting ready enrolled device that nothing holds, never a boot, provision or eviction): grant a fitting ready device, wait for a fitting one the warm pool is booting, boot a fitting shut-down one, evict an idle running device, provision, wait, or refuse. Takes capacity reservations and claims and hands them to the caller. | Perform any side effect. | | `LeaseLifecycle` | `src/leasing/lease-lifecycle.ts` | Grant, renew, the registry half of a release, expiry scheduling through `LeaseExpiryScheduler`. | Reclaim; wake the queue. | | `LeaseExpiryScheduler` | `src/leasing/lease-expiry-scheduler.ts` | TTL timers, delivered to the release coordinator. | Decide what expiry means. | -| `LeaseReleaseCoordinator` | `src/leasing/lease-release-coordinator.ts` | Release, release-all, expiry, device-lost and the startup ending (reclaim, wait with no reclaim started, or missing in one write) as lease commands: the serialized commit, the backgrounded reclaim with its claim, the drain for nuke and daemon stop, maintenance admission. | Purge or decide warmth: it hands the released device to `ReclaimCoordinator`. | +| `LeaseReleaseCoordinator` | `src/leasing/lease-release-coordinator.ts` | Release, release-all, expiry, device-lost and the startup ending (reclaim, wait with no reclaim started, or missing in one write) as lease commands: the serialized commit, the lease-end participants called after it (bounded by `lease.endWaitMs`, cut short by nuke, stop and dispose), the backgrounded reclaim with its claim, the drain for nuke and daemon stop, maintenance admission. | Purge or decide warmth: it hands the released device to `ReclaimCoordinator`. | | `LeaseHealthMonitor` | `src/leasing/lease-health-monitor.ts` | Observing leased devices on a tick, rebooting a crashed one under its lease, giving the lease up as `device-lost` when recovery is exhausted. | Touch an unleased device. | | `PhysicalDeviceWatcher` | `src/leasing/physical-device-watcher.ts` | The presence tick (every `health.probeIntervalMs`, whatever `health.enabled` says), the grace-period count of not-present reads, `absent` and its reason, the device-lost ending of a lease on a device that is gone, and the return reclaim of a device that is back (ADR 0022 §7). | Retry quarantine, grant, or run a driver verb but the presence listing on a leased device. | diff --git a/src/cli/index.test.ts b/src/cli/index.test.ts index fcae31fe..847bcb8c 100644 --- a/src/cli/index.test.ts +++ b/src/cli/index.test.ts @@ -5846,6 +5846,7 @@ function testConfig(): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/contract/operations.test.ts b/src/contract/operations.test.ts index 0f0e47d8..87442773 100644 --- a/src/contract/operations.test.ts +++ b/src/contract/operations.test.ts @@ -1071,6 +1071,7 @@ describe("operation input/output round trips", () => { requestRetentionMs: 1, maxRequestRecords: 1, maxPerRequester: 1, + endWaitMs: 600_000, }, exec: { timeoutMs: 1 }, diskPressure: { freeBytesThreshold: 1 }, @@ -1126,6 +1127,7 @@ describe("operation input/output round trips", () => { requestRetentionMs: 1, maxRequestRecords: 1, maxPerRequester: 1, + endWaitMs: 600_000, }, exec: { timeoutMs: 1 }, diskPressure: { freeBytesThreshold: 1 }, diff --git a/src/contract/protocol.ts b/src/contract/protocol.ts index 94b89a0b..dd7ef7f8 100644 --- a/src/contract/protocol.ts +++ b/src/contract/protocol.ts @@ -10,7 +10,7 @@ export interface ProtocolRange { const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() }); /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */ -export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 28, max: 28 }; +export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 29, max: 29 }; const LEGACY_DAEMON_PROTOCOL_VERSION = 2; diff --git a/src/contract/schemas.ts b/src/contract/schemas.ts index eb05651b..374160ab 100644 --- a/src/contract/schemas.ts +++ b/src/contract/schemas.ts @@ -818,6 +818,7 @@ export const configSchema = z.object({ requestRetentionMs: z.number(), maxRequestRecords: z.number(), maxPerRequester: z.number(), + endWaitMs: z.number(), }), exec: z.object({ timeoutMs: z.number() }), diskPressure: z.object({ freeBytesThreshold: z.number() }), diff --git a/src/core/cleanup/idle-destroy.test.ts b/src/core/cleanup/idle-destroy.test.ts index 45e9ec7d..2bbe9778 100644 --- a/src/core/cleanup/idle-destroy.test.ts +++ b/src/core/cleanup/idle-destroy.test.ts @@ -53,6 +53,7 @@ const config: Config = { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/core/cleanup/idle-shutdown.test.ts b/src/core/cleanup/idle-shutdown.test.ts index 17ebf360..5d6fd88e 100644 --- a/src/core/cleanup/idle-shutdown.test.ts +++ b/src/core/cleanup/idle-shutdown.test.ts @@ -51,6 +51,7 @@ const config: Config = { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/core/config.test.ts b/src/core/config.test.ts index 17bcf78a..4deccb7d 100644 --- a/src/core/config.test.ts +++ b/src/core/config.test.ts @@ -100,6 +100,7 @@ describe("loadConfig", () => { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", @@ -683,6 +684,50 @@ describe("loadConfig", () => { }, ); + it("lease.endWaitMs defaults to 600000 and accepts a positive number and 2147483647", async () => { + const filesystem = new MemoryFilesystem(); + await filesystem.mkdirp("/home/agent/.simlock"); + const unset = await loadConfig({ configPath, filesystem, systemStats: createStats() }); + + await filesystem.writeFileAtomic(configPath, JSON.stringify({ lease: { endWaitMs: 5_000 } })); + const small = await loadConfig({ configPath, filesystem, systemStats: createStats() }); + await filesystem.writeFileAtomic( + configPath, + JSON.stringify({ lease: { endWaitMs: 2_147_483_647 } }), + ); + const largest = await loadConfig({ configPath, filesystem, systemStats: createStats() }); + + expect(unset.lease.endWaitMs).toBe(600_000); + expect(small.lease.endWaitMs).toBe(5_000); + expect(largest.lease.endWaitMs).toBe(2_147_483_647); + }); + + it.each([[0], [-1], [2_147_483_648], ["600000"]])( + "refuses lease.endWaitMs %j with a message naming the key, where 5000 loads", + async (value) => { + const filesystem = new MemoryFilesystem(); + await filesystem.mkdirp("/home/agent/.simlock"); + await filesystem.writeFileAtomic(configPath, JSON.stringify({ lease: { endWaitMs: 5_000 } })); + const accepted = await loadConfig({ configPath, filesystem, systemStats: createStats() }); + expect(accepted.lease.endWaitMs).toBe(5_000); + await filesystem.writeFileAtomic(configPath, JSON.stringify({ lease: { endWaitMs: value } })); + + await expect( + loadConfig({ configPath, filesystem, systemStats: createStats() }), + ).rejects.toThrow("lease.endWaitMs"); + }, + ); + + it("a loaded config parsed through configSchema keeps lease.endWaitMs", async () => { + const config = await loadConfig({ + configPath, + filesystem: new MemoryFilesystem(), + systemStats: createStats(), + }); + + expect((configSchema.parse(config) as unknown as Config).lease.endWaitMs).toBe(600_000); + }); + it("reads lease.identity per platform and leaves the unset platform reusable", async () => { const filesystem = new MemoryFilesystem(); await filesystem.mkdirp("/home/agent/.simlock"); @@ -763,6 +808,7 @@ describe("loadConfig", () => { defaultTtlMs: 15 * 60_000, identity: { android: "reusable", ios: "reusable" }, maxPerRequester: 1, + endWaitMs: 600_000, maxRequestRecords: 10_000, maxTtlMs: 4 * 60 * 60_000, requestRetentionMs: 10 * 60_000, diff --git a/src/core/config.ts b/src/core/config.ts index 67fd0805..00bc45f3 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -181,6 +181,7 @@ export interface Config { */ readonly maxRequestRecords: number; readonly maxPerRequester: number; + readonly endWaitMs: number; }; /** * ADR 0005 §19e. Platform-agnostic on purpose: it bounds the *daemon's* willingness to wait @@ -678,6 +679,7 @@ function defaultConfig( requestRetentionMs: DEFAULT_LEASE_REQUEST_RETENTION_MS, maxRequestRecords: DEFAULT_LEASE_MAX_REQUEST_RECORDS, maxPerRequester: DEFAULT_LEASE_MAX_PER_REQUESTER, + endWaitMs: DEFAULT_EXEC_TIMEOUT_MS, }, exec: { timeoutMs: DEFAULT_EXEC_TIMEOUT_MS }, diskPressure: { freeBytesThreshold: 10 * 1024 ** 3 }, @@ -827,6 +829,7 @@ function configValidators(strategy: CapacityStrategyName): Record>; +/** why: `setTimeout` fires after about 1 ms above this delay, which would end the wait at once. */ +const MAXIMUM_TIMER_DELAY_MS = 2_147_483_647; + +const endWaitValidator: Validator = (value, path) => { + const wait = positiveNumber(value, path); + if (wait > MAXIMUM_TIMER_DELAY_MS) { + throw invalidValue(path, `a positive number no larger than ${MAXIMUM_TIMER_DELAY_MS}`); + } + return wait; +}; + /** * Each key is a device class and each value one non-empty model name or a non-empty list of * them, stored as a list. Any other key is refused rather than warned about: a misspelled class diff --git a/src/core/doctor.test.ts b/src/core/doctor.test.ts index 000c2fa0..3510c49d 100644 --- a/src/core/doctor.test.ts +++ b/src/core/doctor.test.ts @@ -3052,6 +3052,7 @@ function config(stalledTransitionOverrides: Partial requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/core/nuke.test.ts b/src/core/nuke.test.ts index d75bd0c4..3153ca02 100644 --- a/src/core/nuke.test.ts +++ b/src/core/nuke.test.ts @@ -191,6 +191,7 @@ function config(): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/core/reaper.test.ts b/src/core/reaper.test.ts index 1cfe6940..600fe63b 100644 --- a/src/core/reaper.test.ts +++ b/src/core/reaper.test.ts @@ -87,6 +87,7 @@ function config(): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/daemon-protocol/index.test.ts b/src/daemon-protocol/index.test.ts index fe2a770c..6b370620 100644 --- a/src/daemon-protocol/index.test.ts +++ b/src/daemon-protocol/index.test.ts @@ -9,7 +9,7 @@ import { describe("daemon protocol", () => { it("keeps the current protocol version and newline framing", () => { - expect(DAEMON_PROTOCOL_VERSION).toBe(28); + expect(DAEMON_PROTOCOL_VERSION).toBe(29); expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n'); }); diff --git a/src/daemon/dispatcher.test.ts b/src/daemon/dispatcher.test.ts index 8417e728..ee2bf2f6 100644 --- a/src/daemon/dispatcher.test.ts +++ b/src/daemon/dispatcher.test.ts @@ -3581,6 +3581,7 @@ function testConfig( requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, ...leaseOverrides, }, capacity, diff --git a/src/daemon/server.test.ts b/src/daemon/server.test.ts index 0b822790..7687ec21 100644 --- a/src/daemon/server.test.ts +++ b/src/daemon/server.test.ts @@ -190,8 +190,9 @@ describe("DaemonServer", () => { [24, "cannot send lease.request's physical or read physicalQueueDepth"], [26, "cannot read a physical device's presence"], [27, "cannot read config.get's lease.maxPerRequester"], + [28, "cannot read config.get's lease.endWaitMs"], ])( - "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 28, to a client on protocol %i, which %s", + "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 29, to a client on protocol %i, which %s", async (version) => { const harness = await createHarness(); const previous = await createClient(harness.socketPath); @@ -201,7 +202,7 @@ describe("DaemonServer", () => { ).resolves.toMatchObject({ error: { code: "PROTOCOL_VERSION_UNSUPPORTED", - details: { client: { min: version, max: version }, daemon: { min: 28, max: 28 } }, + details: { client: { min: version, max: version }, daemon: { min: 29, max: 29 } }, }, ok: false, }); @@ -3438,6 +3439,7 @@ function testConfig( requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, ...leaseOverrides, }, capacity: { diff --git a/src/gateway/dispatcher.test.ts b/src/gateway/dispatcher.test.ts index f32c5208..8bd33e50 100644 --- a/src/gateway/dispatcher.test.ts +++ b/src/gateway/dispatcher.test.ts @@ -62,6 +62,7 @@ const gatewayConfig = { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, log: { level: "info" as const, rotateBytes: 1 }, mode: "gateway" as const, diff --git a/src/gateway/service.test.ts b/src/gateway/service.test.ts index 4dbbe2a2..2cd948d1 100644 --- a/src/gateway/service.test.ts +++ b/src/gateway/service.test.ts @@ -1207,7 +1207,7 @@ describe("GatewayService", () => { min: PROTOCOL_VERSION_RANGE.max - 1, max: PROTOCOL_VERSION_RANGE.max - 1, }; - expect(PROTOCOL_VERSION_RANGE.max).toBe(28); + expect(PROTOCOL_VERSION_RANGE.max).toBe(29); expect(negotiateProtocolVersion(PROTOCOL_VERSION_RANGE, previous)).toBeUndefined(); const harness = fleet(); await harness.service.start(); diff --git a/src/http/test-fakes.ts b/src/http/test-fakes.ts index 548c4b30..522f22c6 100644 --- a/src/http/test-fakes.ts +++ b/src/http/test-fakes.ts @@ -76,6 +76,7 @@ export function testConfig( requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, ...overrides, }, log: { level: "info", rotateBytes: 5 * 1024 * 1024 }, diff --git a/src/leasing/acquisition-planner.test.ts b/src/leasing/acquisition-planner.test.ts index 8b2820fb..849f2b79 100644 --- a/src/leasing/acquisition-planner.test.ts +++ b/src/leasing/acquisition-planner.test.ts @@ -62,6 +62,7 @@ const config: Config = { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/leasing/create-leasing.test.ts b/src/leasing/create-leasing.test.ts index 1d2f3421..9dca1a29 100644 --- a/src/leasing/create-leasing.test.ts +++ b/src/leasing/create-leasing.test.ts @@ -79,6 +79,7 @@ function config( requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, ...overrides, }, capacity: { diff --git a/src/leasing/create-leasing.ts b/src/leasing/create-leasing.ts index 8938ba0b..720a2b16 100644 --- a/src/leasing/create-leasing.ts +++ b/src/leasing/create-leasing.ts @@ -24,6 +24,7 @@ import type { ClientReleaseReason, DeviceModeReader, LeaseCommands, + LeaseEndParticipant, QueueControl, } from "./lease-ports.js"; import { LeaseReleaseCoordinator } from "./lease-release-coordinator.js"; @@ -55,6 +56,7 @@ export interface LeasingOptions { readonly modelPreferences?: ModelPreferences | undefined; /** why: not started on build: a health probe would shell out while startup convergence does. */ readonly healthMonitor?: boolean | undefined; + readonly leaseEndParticipants?: readonly LeaseEndParticipant[] | undefined; } export interface Leasing extends LeaseCommands, QueueControl, DeviceModeReader { @@ -160,6 +162,15 @@ export function createLeasing(options: LeasingOptions): Leasing { const releaseCoordinator: LeaseReleaseCoordinator = new LeaseReleaseCoordinator({ claims: core.claims, decisions, + ...(options.leaseEndParticipants === undefined || options.leaseEndParticipants.length === 0 + ? {} + : { + leaseEnd: { + clock: options.clock, + endWaitMs: options.config.lease.endWaitMs, + participants: options.leaseEndParticipants, + }, + }), lifecycle: leases, ...(options.logger === undefined ? {} : { logger: options.logger }), // Called once a backgrounded reclaim has given up its claim; the warm pool can act on the @@ -256,6 +267,7 @@ export function createLeasing(options: LeasingOptions): Leasing { await releaseCoordinator.settleBackgroundReclaims(); }, dispose: () => { + releaseCoordinator.dispose(); physicalDeviceWatcher.dispose(); expiry.dispose(); }, diff --git a/src/leasing/index.ts b/src/leasing/index.ts index 52656b7c..cd17e457 100644 --- a/src/leasing/index.ts +++ b/src/leasing/index.ts @@ -2,7 +2,15 @@ export { createLeasing } from "./create-leasing.js"; export { NoCapacityError } from "./lease-acquisition-coordinator.js"; export { LeaseHealthMonitor } from "./lease-health-monitor.js"; export { type PhysicalDeviceWatcher } from "./physical-device-watcher.js"; -export { type DeviceModeReader, type LeaseCommands, type QueueControl } from "./lease-ports.js"; +export { + type DeviceModeReader, + // fallow-ignore-next-line unused-type -- the port for optional modules (#474 registers the first participant). + type EndedLease, + type LeaseCommands, + // fallow-ignore-next-line unused-type -- the port for optional modules (#474 registers the first participant). + type LeaseEndParticipant, + type QueueControl, +} from "./lease-ports.js"; export { IdempotencyConflictError, InMemoryLeaseRequestStore, diff --git a/src/leasing/lease-acquisition-coordinator.test.ts b/src/leasing/lease-acquisition-coordinator.test.ts index 74e5a2da..a04d42b7 100644 --- a/src/leasing/lease-acquisition-coordinator.test.ts +++ b/src/leasing/lease-acquisition-coordinator.test.ts @@ -97,6 +97,7 @@ function config(maxDevices = 1, maxRunning = 1): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/leasing/lease-end-participants.test.ts b/src/leasing/lease-end-participants.test.ts new file mode 100644 index 00000000..faca01d9 --- /dev/null +++ b/src/leasing/lease-end-participants.test.ts @@ -0,0 +1,1069 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { EventBus } from "../bus/index.js"; +import { + type Config, + type DeviceRequest, + type DriverDevice, + type LeaseGrant, + Registry, + UnknownLeaseError, +} from "../core/index.js"; +import { FakeDriver, type FakeDriverOptions, testComponentWiring } from "../core/testing.js"; +import { + FakeClock, + FakeSystemStats, + JsonLinesLogger, + MemoryFilesystem, + MemoryLogSink, +} from "../ports/index.js"; +import { NoCapacityError } from "./lease-acquisition-coordinator.js"; +import type { EndedLease, LeaseEndParticipant } from "./lease-ports.js"; +import { createTestEngine, type TestEngine } from "./testing.js"; + +const gibibyte = 1024 ** 3; +const statePath = "/home/agent/.simlock/state.json"; +const END_WAIT_MS = 600_000; +const PROBE_MS = 30_000; + +const request = { model: "iPhone 16", osVersion: "26.5", platform: "ios" } as const; +const otherRequest = { model: "iPhone 15", osVersion: "26.5", platform: "ios" } as const; +const physicalRequest = { + model: "iPhone 15 Pro", + physical: true, + platform: "ios", +} as const satisfies DeviceRequest; + +interface Observed { + readonly claimed: boolean; + readonly deviceState: string | undefined; + readonly events: readonly string[]; + readonly leaseCount: number; + readonly marker: string | undefined; +} + +interface RecordedCall { + readonly abortedAtCall: boolean; + readonly ended: EndedLease; + readonly observed: Observed | undefined; + readonly signal: AbortSignal; + reject(error: Error): void; + resolve(): void; +} + +class TestParticipant implements LeaseEndParticipant { + readonly calls: RecordedCall[] = []; + observer: ((ended: EndedLease) => Observed) | undefined; + + leaseEnded(ended: EndedLease, signal: AbortSignal): Promise { + const observed = this.observer?.(ended); + return new Promise((resolve, reject) => { + this.calls.push({ + abortedAtCall: signal.aborted, + ended, + observed, + reject, + resolve, + signal, + }); + }); + } +} + +class ThrowingParticipant implements LeaseEndParticipant { + leaseEnded(): Promise { + throw new Error("participant threw"); + } +} + +interface World { + readonly bus: EventBus; + readonly clock: FakeClock; + readonly driver: FakeDriver; + readonly filesystem: MemoryFilesystem; + readonly ids: { next: number }; + readonly sink: MemoryLogSink; +} + +function createWorld(driverOptions: Partial = {}): World { + const clock = new FakeClock(1_000); + return { + bus: new EventBus(clock), + clock, + driver: new FakeDriver({ + availableOsVersions: ["26.5"], + clock, + platform: "ios", + ...driverOptions, + }), + filesystem: new MemoryFilesystem(), + ids: { next: 1 }, + sink: new MemoryLogSink(), + }; +} + +interface EngineOptions { + readonly ios?: { readonly maxDevices: number; readonly maxRunning: number }; + readonly lease?: Partial; + readonly participants?: readonly LeaseEndParticipant[]; + readonly world?: World; +} + +function config(options: EngineOptions): Config { + const ios = options.ios ?? { maxDevices: 1, maxRunning: 1 }; + return { + mode: "worker", + exec: { timeoutMs: 600_000 }, + diskPressure: { freeBytesThreshold: 10 * gibibyte }, + gateway: { + disconnectedRetentionMs: 24 * 60 * 60_000, + execTimeoutMs: 11 * 60_000, + leaseRequestTimeoutMs: 5 * 60_000, + routing: "warm-then-free" as const, + }, + drivers: {}, + downloads: { acceptAndroidLicenses: false, policy: "on-request", timeoutMs: 1_200_000 }, + eventBuffer: { capacity: 1_000 }, + http: { enabled: false, host: "127.0.0.1", port: 4700 }, + ios: { defaultMode: "full", defaultModels: {}, slim: { bootTimeoutMs: 600_000 } }, + android: { + defaultModels: {}, + emulator: { headless: false, gpu: "auto", audio: true, bootAnimation: true }, + }, + health: { + enabled: true, + maxConcurrentRecoveries: 1, + maxRecoveryAttempts: 3, + probeIntervalMs: PROBE_MS, + recoveryBackoffMs: 5_000, + stableObservations: 2, + }, + stalledTransition: { thresholdMultiplier: 3, minimumThresholdMs: 60_000 }, + idle: { deleteAfterMs: 48 * 60 * 60_000, shutdownAfterMs: 24 * 60 * 60_000 }, + lease: { + defaultTtlMs: 3_600_000, + maxTtlMs: 14_400_000, + identity: { ios: "reusable", android: "reusable" }, + requestRetentionMs: 600_000, + maxRequestRecords: 10_000, + maxPerRequester: 1, + endWaitMs: END_WAIT_MS, + ...options.lease, + }, + capacity: { + strategy: "resource", + config: { + limits: { + android: { maxDevices: 1, maxRunning: 1 }, + ios, + maxRunning: ios.maxRunning + 1, + }, + ramBudget: { androidBytesPerDevice: 4 * gibibyte, iosBytesPerDevice: gibibyte }, + }, + }, + log: { level: "info", rotateBytes: 5 * 1024 * 1024 }, + eventLog: { + rotateBytes: 5 * 1024 * 1024, + retention: 7 * 24 * 60 * 60 * 1000, + maxBytes: 256 * 1024 * 1024, + }, + warmPool: { + enabled: true, + maxConcurrentBoots: 1, + reserveRunning: { android: 0, ios: 0 }, + targets: [], + quarantine: { + maxRetries: 3, + maxRetryBackoffMs: 300_000, + retryBackoffMs: 30_000, + retryBackoffMultiplier: 2, + }, + }, + }; +} + +interface Harness extends World { + readonly engine: TestEngine; + readonly registry: Registry; + readonly world: World; +} + +const started: TestEngine[] = []; + +afterEach(() => { + for (const engine of started.splice(0)) engine.dispose(); +}); + +async function startEngine(options: EngineOptions = {}): Promise { + const world = options.world ?? createWorld(); + const { bus, clock, driver, filesystem, ids, sink } = world; + const registry = await Registry.load({ + clock, + eventBus: bus, + filesystem, + idGenerator: { generate: () => `${ids.next++}` }, + statePath, + }); + const engine = createTestEngine({ + ...testComponentWiring({ clock, drivers: [driver], eventBus: bus, registry }), + clock, + config: config(options), + drivers: [driver], + eventBus: bus, + idGenerator: { generate: () => `request-${ids.next++}` }, + ...(options.participants === undefined ? {} : { leaseEndParticipants: options.participants }), + logger: new JsonLinesLogger({ clock, level: "debug", sink }), + registry, + systemStats: new FakeSystemStats({ cpuCount: 8, totalRamBytes: 32 * gibibyte }), + }); + started.push(engine); + for (const participant of options.participants ?? []) { + if (participant instanceof TestParticipant) { + participant.observer = (ended) => { + const device = registry.snapshot.devices.find( + (candidate) => candidate.id === ended.deviceId, + ); + return { + claimed: engine.claimReader.isClaimed(ended.deviceId), + deviceState: device?.state, + events: bus.replay().map((event) => String(event.event)), + leaseCount: registry.snapshot.leases.length, + marker: (device as { deferredReclaimLeaseId?: string } | undefined) + ?.deferredReclaimLeaseId, + }; + }; + } + } + return { ...world, engine, registry, world }; +} + +function asking(requesterId: string) { + return { ownerId: `${requesterId}-owner`, requesterId }; +} + +async function flush(): Promise { + for (let count = 0; count < 500; count += 1) { + await Promise.resolve(); + } +} + +async function settledOrPending(promise: Promise): Promise { + return Promise.race([ + promise.catch((error: unknown) => error), + flush().then(() => "still pending"), + ]); +} + +function reclaims(driver: FakeDriver) { + return driver.calls.filter((call) => call.operation === "reclaim"); +} + +function stateOf(harness: Harness, deviceId: string): string | undefined { + return harness.registry.snapshot.devices.find((device) => device.id === deviceId)?.state; +} + +function markerOf(harness: Harness, deviceId: string): string | undefined { + const device = harness.registry.snapshot.devices.find((candidate) => candidate.id === deviceId); + return (device as { deferredReclaimLeaseId?: string } | undefined)?.deferredReclaimLeaseId; +} + +function warningsFor(world: World, leaseId: string) { + return world.sink.records.filter( + (record) => record.level === "warn" && record.fields?.leaseId === leaseId, + ); +} + +function errorsFor(world: World, leaseId: string) { + return world.sink.records.filter( + (record) => record.level === "error" && record.fields?.leaseId === leaseId, + ); +} + +function eventNames(world: World): string[] { + return world.bus.replay().map((event) => String(event.event)); +} + +async function withoutUnhandledRejections(run: () => Promise): Promise { + const seen: unknown[] = []; + const listener = (reason: unknown): void => { + seen.push(reason); + }; + process.on("unhandledRejection", listener); + try { + await run(); + await new Promise((resolve) => setImmediate(resolve)); + } finally { + process.off("unhandledRejection", listener); + } + expect(seen).toEqual([]); +} + +async function leaseAndRelease(options: EngineOptions = {}) { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant], ...options }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + return { grant, harness, participant }; +} + +async function endStartupLease(options: { + readonly afterFirstStop?: (first: Harness, grant: LeaseGrant) => void | Promise; + readonly participant: TestParticipant; + readonly ttlMs?: number; +}) { + const world = createWorld(); + const first = await startEngine({ world }); + const grant = await first.engine.request(request, { + ...asking("agent"), + ...(options.ttlMs === undefined ? {} : { ttlMs: options.ttlMs }), + }); + first.engine.dispose(); + await options.afterFirstStop?.(first, grant); + const second = await startEngine({ participants: [options.participant], world }); + return { grant, second, world }; +} + +function driverDeviceOf(harness: Harness, deviceId: string): DriverDevice { + const device = harness.registry.snapshot.devices.find((candidate) => candidate.id === deviceId); + if (device === undefined) throw new Error(`no device ${deviceId}`); + return { address: "", deviceId: device.driverDeviceId, driverData: device.driverData }; +} + +describe("lease-end participants (#469)", () => { + it("a release answers its caller while the participant has not returned, and the participant gets the lease id, owner, simlock device id, platform and reason explicit", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + const grant = await harness.engine.request(request, asking("agent")); + + const released = harness.engine.release(grant.lease.id, "explicit"); + + await expect(settledOrPending(released)).resolves.toBeUndefined(); + await flush(); + expect(participant.calls.map((call) => call.ended)).toEqual([ + { + deviceId: grant.device.id, + leaseId: grant.lease.id, + ownerId: "agent-owner", + platform: "ios", + reason: "explicit", + }, + ]); + }); + + it("the participant is called after lease.released is emitted, while the device is reclaiming and has no lease", async () => { + const { grant, participant } = await leaseAndRelease(); + + expect(participant.calls).toHaveLength(1); + expect(participant.calls[0]?.observed).toMatchObject({ + deviceState: "reclaiming", + leaseCount: 0, + }); + expect(participant.calls[0]?.observed?.events).toContain("lease.released"); + expect(grant.device.id).toBe(participant.calls[0]?.ended.deviceId); + }); + + it("the driver's reclaim does not start until the participant returns; when it returns, the reclaim runs and the device becomes ready", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + expect(reclaims(harness.driver)).toHaveLength(0); + expect(stateOf(harness, grant.device.id)).toBe("reclaiming"); + + participant.calls[0]?.resolve(); + await flush(); + + expect(reclaims(harness.driver)).toHaveLength(1); + expect(stateOf(harness, grant.device.id)).toBe("ready"); + }); + + it("while the participant has not returned, a request for the same spec is not granted that device, and it is granted it after the reclaim", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + + const second = harness.engine.request(request, asking("next")); + + await expect(settledOrPending(second)).resolves.toBe("still pending"); + participant.calls[0]?.resolve(); + await flush(); + await harness.engine.settle(); + await expect(settledOrPending(second)).resolves.toMatchObject({ + device: { id: grant.device.id }, + }); + }); + + it("the device is claimed from before the participant is called until the reclaim ends", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + + expect(participant.calls[0]?.observed?.claimed).toBe(true); + expect(harness.engine.claimReader.isClaimed(grant.device.id)).toBe(true); + participant.calls[0]?.resolve(); + await flush(); + await harness.engine.settle(); + expect(reclaims(harness.driver)).toHaveLength(1); + expect(harness.engine.claimReader.isClaimed(grant.device.id)).toBe(false); + }); + + it("while the participant has not returned, a renew of another lease and a grant of another device both complete", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ + ios: { maxDevices: 3, maxRunning: 3 }, + participants: [participant], + }); + const ending = await harness.engine.request(request, asking("ending")); + const other = await harness.engine.request(request, asking("other")); + await harness.engine.release(ending.lease.id, "explicit"); + await flush(); + expect(participant.calls).toHaveLength(1); + + const renewed = harness.engine.renew(other.lease.id, 120_000); + const granted = harness.engine.request(request, asking("third")); + + await expect(settledOrPending(renewed)).resolves.toMatchObject({ id: other.lease.id }); + const third = await settledOrPending(granted); + expect(third).toMatchObject({ lease: { requesterId: "third" } }); + expect((third as LeaseGrant).device.id).not.toBe(ending.device.id); + }); + + it("a participant that never returns: 1 ms before lease.endWaitMs the reclaim has not started; at lease.endWaitMs its signal is aborted, a warning naming the lease and device is logged, and the reclaim starts; when that participant resolves later, the driver's reclaim has still run once", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + + harness.clock.advance(END_WAIT_MS - 1); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(0); + expect(participant.calls[0]?.signal.aborted).toBe(false); + + harness.clock.advance(1); + await flush(); + expect(participant.calls[0]?.signal.aborted).toBe(true); + expect(warningsFor(harness, grant.lease.id)).toMatchObject([ + { fields: { deviceId: grant.device.id, leaseId: grant.lease.id } }, + ]); + expect(reclaims(harness.driver)).toHaveLength(1); + + participant.calls[0]?.resolve(); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("two participants are both called before either returns; the reclaim waits for the slower one", async () => { + const fast = new TestParticipant(); + const slow = new TestParticipant(); + const harness = await startEngine({ participants: [fast, slow] }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + expect(fast.calls).toHaveLength(1); + expect(slow.calls).toHaveLength(1); + + fast.calls[0]?.resolve(); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(0); + + slow.calls[0]?.resolve(); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("one ceiling, counted from the call, covers both participants and is not reset when the first returns", async () => { + const fast = new TestParticipant(); + const slow = new TestParticipant(); + const harness = await startEngine({ participants: [fast, slow] }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + expect(fast.calls).toHaveLength(1); + + harness.clock.advance(END_WAIT_MS / 2); + fast.calls[0]?.resolve(); + await flush(); + harness.clock.advance(END_WAIT_MS / 2 - 1); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(0); + + harness.clock.advance(1); + await flush(); + expect(slow.calls[0]?.signal.aborted).toBe(true); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("a participant that rejects, and one that throws, are each logged as an error with the lease id, device id and error message; the other participant is still awaited; the device becomes ready", async () => { + const rejecting = new TestParticipant(); + const waiting = new TestParticipant(); + const harness = await startEngine({ + participants: [rejecting, new ThrowingParticipant(), waiting], + }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + expect(rejecting.calls).toHaveLength(1); + expect(waiting.calls).toHaveLength(1); + + rejecting.calls[0]?.reject(new Error("participant rejected")); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(0); + expect(errorsFor(harness, grant.lease.id)).toMatchObject([ + { fields: { deviceId: grant.device.id, error: "participant threw" } }, + { fields: { deviceId: grant.device.id, error: "participant rejected" } }, + ]); + + waiting.calls[0]?.resolve(); + await flush(); + expect(stateOf(harness, grant.device.id)).toBe("ready"); + }); + + it("a participant that rejects after the ceiling passed is logged as an error with the lease id; the run reports no unhandled rejection", async () => { + await withoutUnhandledRejections(async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + harness.clock.advance(END_WAIT_MS); + await flush(); + expect(reclaims(harness.driver)).toHaveLength(1); + + participant.calls[0]?.reject(new Error("rejected late")); + await flush(); + + expect(errorsFor(harness, grant.lease.id)).toMatchObject([ + { fields: { error: "rejected late" } }, + ]); + }); + }); + + it("a participant that rejects after nuke cut its wait short is logged as an error with the lease id; the run reports no unhandled rejection", async () => { + await withoutUnhandledRejections(async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + await harness.engine.nuke(false); + + participant.calls[0]?.reject(new Error("rejected after nuke")); + await flush(); + + expect(errorsFor(harness, grant.lease.id)).toMatchObject([ + { fields: { error: "rejected after nuke" } }, + ]); + }); + }); + + it.each(["explicit", "killed"] as const)( + "release with reason %s calls the participant with that reason before the reclaim", + async (reason) => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + const grant = await harness.engine.request(request, asking("agent")); + + await harness.engine.release(grant.lease.id, reason); + await flush(); + + expect(participant.calls.map((call) => call.ended.reason)).toEqual([reason]); + expect(reclaims(harness.driver)).toHaveLength(0); + }, + ); + + it("releaseAll calls the participant once per lease, before any reclaim", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ + ios: { maxDevices: 2, maxRunning: 2 }, + participants: [participant], + }); + const first = await harness.engine.request(request, asking("one")); + const second = await harness.engine.request(request, asking("two")); + + await harness.engine.releaseAll("explicit"); + await flush(); + + expect(participant.calls.map((call) => call.ended.leaseId).sort()).toEqual( + [first.lease.id, second.lease.id].sort(), + ); + expect(reclaims(harness.driver)).toHaveLength(0); + }); + + it("an expiry calls the participant with reason expired before the reclaim", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ lease: { defaultTtlMs: 10 }, participants: [participant] }); + await harness.engine.request(request, asking("agent")); + + harness.clock.advance(10); + await flush(); + + expect(participant.calls.map((call) => call.ended.reason)).toEqual(["expired"]); + expect(reclaims(harness.driver)).toHaveLength(0); + }); + + it("the health monitor's device-lost release calls the participant with reason device-lost before the reclaim", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + await harness.engine.request(request, asking("agent")); + harness.driver.setManagedReality({ devices: [], processes: [] }); + + harness.engine.healthMonitor.start(); + for (let tick = 0; tick < 3; tick += 1) { + harness.clock.advance(PROBE_MS); + await flush(); + } + harness.engine.healthMonitor.dispose(); + + expect(participant.calls.map((call) => call.ended.reason)).toEqual(["device-lost"]); + expect(reclaims(harness.driver)).toHaveLength(0); + }); + + it.each([ + ["expired", 10_000], + ["device-lost", 3_600_000], + ] as const)( + "a startup ending with reclaim calls the participant with reason %s before the reclaim", + async (reason, ttlMs) => { + const participant = new TestParticipant(); + const { grant, second, world } = await endStartupLease({ + afterFirstStop: async (first, leased) => { + await first.driver.shutdown(driverDeviceOf(first, leased.device.id)); + first.clock.advance(20_000); + }, + participant, + ttlMs, + }); + + await second.engine.convergeRunningCapacity(); + await flush(); + + expect(participant.calls.map((call) => call.ended)).toMatchObject([ + { deviceId: grant.device.id, leaseId: grant.lease.id, reason }, + ]); + expect(reclaims(world.driver)).toHaveLength(0); + }, + ); + + it("a startup ending with reclaim whose participant has not returned, followed by core.converge: converge leaves the device alone, and the device is wiped once the participant returns", async () => { + const participant = new TestParticipant(); + const { grant, second, world } = await endStartupLease({ + afterFirstStop: async (first, leased) => { + await first.driver.shutdown(driverDeviceOf(first, leased.device.id)); + }, + participant, + }); + const callsBefore = world.driver.calls.length; + + await second.engine.convergeRunningCapacity(); + await flush(); + + expect(participant.calls).toHaveLength(1); + expect( + world.driver.calls.slice(callsBefore).filter((call) => call.operation === "shutdown"), + ).toEqual([]); + expect( + world.bus + .replay() + .filter( + (event) => + String(event.event) === "device.shutdown" && + (event.payload as { initiator?: string }).initiator === "startup-interrupted-reclaim", + ), + ).toEqual([]); + expect(reclaims(world.driver)).toHaveLength(0); + + participant.calls[0]?.resolve(); + await flush(); + await second.engine.settle(); + expect(reclaims(world.driver)).toHaveLength(1); + expect(stateOf(second, grant.device.id)).toBe("ready"); + }); + + it("a startup ending with wait calls the participant with its reason, starts no reclaim, and leaves the device reclaiming with its deferred wipe recorded", async () => { + const participant = new TestParticipant(); + const { grant, second, world } = await endStartupLease({ + afterFirstStop: (first) => { + first.driver.failOn( + "listManaged", + first.driver.calls.filter((call) => call.operation === "listManaged").length + 1, + new Error("cannot list"), + ); + }, + participant, + }); + + await second.engine.convergeRunningCapacity(); + await flush(); + participant.calls[0]?.resolve(); + await flush(); + + expect(participant.calls.map((call) => call.ended)).toMatchObject([ + { deviceId: grant.device.id, leaseId: grant.lease.id, reason: "device-lost" }, + ]); + expect(reclaims(world.driver)).toHaveLength(0); + expect(stateOf(second, grant.device.id)).toBe("reclaiming"); + expect(markerOf(second, grant.device.id)).toBe(grant.lease.id); + }); + + it("a startup ending with missing calls the participant and marks the device missing in the same write as today, without waiting for it", async () => { + const participant = new TestParticipant(); + const { grant, second } = await endStartupLease({ + afterFirstStop: (first) => { + first.driver.setManagedReality({ devices: [], processes: [] }); + }, + participant, + }); + + await expect( + settledOrPending(second.engine.convergeRunningCapacity()), + ).resolves.toBeUndefined(); + + expect(participant.calls.map((call) => call.ended)).toMatchObject([ + { deviceId: grant.device.id, leaseId: grant.lease.id, reason: "device-lost" }, + ]); + expect(stateOf(second, grant.device.id)).toBe("deleted"); + expect(second.registry.snapshot.leases).toEqual([]); + }); + + it("a physical device that goes absent (through PhysicalDeviceWatcher) calls the participant with device-lost, and the device becomes absent without waiting for it", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + const device = await harness.registry.enrollPhysical({ + apps: [], + class: "phone", + driverDeviceId: "UDID-A", + model: "iPhone 15 Pro", + osVersion: "18.4", + platform: "ios", + }); + harness.driver.setPhysicalPresence("UDID-A", true); + ( + harness.engine.leasing as unknown as { physicalDeviceWatcher: { start(): void } } + ).physicalDeviceWatcher.start(); + await harness.engine.request(physicalRequest, asking("agent")); + + harness.driver.setPhysicalPresence("UDID-A", false); + for (let tick = 0; tick < 3; tick += 1) { + harness.clock.advance(PROBE_MS); + await flush(); + } + + expect(participant.calls.map((call) => call.ended)).toMatchObject([ + { deviceId: device.id, platform: "ios", reason: "device-lost" }, + ]); + expect(stateOf(harness, device.id)).toBe("absent"); + }); + + it("a call nothing waits for, still running at the ceiling, has its signal aborted then", async () => { + const participant = new TestParticipant(); + const { second } = await endStartupLease({ + afterFirstStop: (first) => { + first.driver.failOn( + "listManaged", + first.driver.calls.filter((call) => call.operation === "listManaged").length + 1, + new Error("cannot list"), + ); + }, + participant, + }); + await second.engine.convergeRunningCapacity(); + await flush(); + expect(participant.calls).toHaveLength(1); + + second.clock.advance(END_WAIT_MS - 1); + await flush(); + expect(participant.calls[0]?.signal.aborted).toBe(false); + + second.clock.advance(1); + await flush(); + expect(participant.calls[0]?.signal.aborted).toBe(true); + }); + + it("a call nothing waits for that rejects is logged as an error with the lease id", async () => { + await withoutUnhandledRejections(async () => { + const participant = new TestParticipant(); + const { grant, second, world } = await endStartupLease({ + afterFirstStop: (first) => { + first.driver.failOn( + "listManaged", + first.driver.calls.filter((call) => call.operation === "listManaged").length + 1, + new Error("cannot list"), + ); + }, + participant, + }); + await second.engine.convergeRunningCapacity(); + await flush(); + expect(participant.calls).toHaveLength(1); + + participant.calls[0]?.reject(new Error("rejected unwaited")); + await flush(); + + expect(errorsFor(world, grant.lease.id)).toMatchObject([ + { fields: { error: "rejected unwaited" } }, + ]); + }); + }); + + it("a release of an unknown lease calls no participant", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + await harness.engine.request(request, asking("agent")); + + await expect(harness.engine.release("no-such-lease", "explicit")).rejects.toBeInstanceOf( + UnknownLeaseError, + ); + await flush(); + + expect(participant.calls).toEqual([]); + }); + + it("an expiry whose deadline was moved by a renew calls no participant", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ + lease: { defaultTtlMs: 100 }, + participants: [participant], + }); + const grant = await harness.engine.request(request, asking("agent")); + harness.clock.advance(90); + await harness.engine.renew(grant.lease.id, 100); + + harness.clock.advance(20); + await flush(); + + expect(harness.registry.snapshot.leases).toHaveLength(1); + expect(participant.calls).toEqual([]); + }); + + it("nuke while a participant has not returned aborts its signal, starts the reclaim at once, and completes without the clock advancing", async () => { + const { harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + + const nuking = harness.engine.nuke(false); + + await expect(settledOrPending(nuking)).resolves.not.toBe("still pending"); + expect(participant.calls[0]?.signal.aborted).toBe(true); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("advancing the clock past lease.endWaitMs after a nuke cut a wait short, and the participant then resolving, logs no ceiling warning and starts no second reclaim", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + await harness.engine.nuke(false); + expect(reclaims(harness.driver)).toHaveLength(1); + + harness.clock.advance(END_WAIT_MS + 1); + participant.calls[0]?.resolve(); + await flush(); + + expect(warningsFor(harness, grant.lease.id)).toEqual([]); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("nuke with leased devices calls each participant with an already aborted signal, and completes without the clock advancing even when the participant never returns", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ + ios: { maxDevices: 2, maxRunning: 2 }, + participants: [participant], + }); + await harness.engine.request(request, asking("one")); + await harness.engine.request(request, asking("two")); + + const nuking = harness.engine.nuke(false); + + await expect(settledOrPending(nuking)).resolves.not.toBe("still pending"); + expect(participant.calls.map((call) => call.abortedAtCall)).toEqual([true, true]); + }); + + it("advancing the clock past lease.endWaitMs after a nuke with leased devices logs no ceiling warning", async () => { + const { grant, harness, participant } = await (async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.nuke(false); + return { grant, harness, participant }; + })(); + expect(participant.calls).toHaveLength(1); + + harness.clock.advance(END_WAIT_MS + 1); + await flush(); + + expect(warningsFor(harness, grant.lease.id)).toEqual([]); + }); + + it("after nuke ends, a release waits for its participant again", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + await harness.engine.nuke(false); + const grant = await harness.engine.request(request, asking("agent")); + + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + + expect(participant.calls).toHaveLength(1); + expect(participant.calls[0]?.abortedAtCall).toBe(false); + expect(reclaims(harness.driver)).toHaveLength(0); + }); + + it("settle() while a participant has not returned aborts its signal and returns once the reclaim settles, without the clock advancing", async () => { + const { harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + + const settling = harness.engine.settle(); + + await expect(settledOrPending(settling)).resolves.not.toBe("still pending"); + expect(participant.calls[0]?.signal.aborted).toBe(true); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("when the participant resolves after settle(), the driver's reclaim has run once, and advancing past lease.endWaitMs logs no ceiling warning", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + await harness.engine.settle(); + + participant.calls[0]?.resolve(); + harness.clock.advance(END_WAIT_MS + 1); + await flush(); + + expect(reclaims(harness.driver)).toHaveLength(1); + expect(warningsFor(harness, grant.lease.id)).toEqual([]); + }); + + it("a release after settle() began calls the participant with an already aborted signal and reclaims at once", async () => { + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant] }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.settle(); + + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + + expect(participant.calls.map((call) => call.abortedAtCall)).toEqual([true]); + expect(reclaims(harness.driver)).toHaveLength(1); + }); + + it("a participant that returns before the ceiling: advancing the clock past lease.endWaitMs afterwards logs no ceiling warning and aborts no signal", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + participant.calls[0]?.resolve(); + await flush(); + + harness.clock.advance(END_WAIT_MS + 1); + await flush(); + + expect(reclaims(harness.driver)).toHaveLength(1); + expect(warningsFor(harness, grant.lease.id)).toEqual([]); + expect(participant.calls[0]?.signal.aborted).toBe(false); + }); + + it("dispose() with a participant still running aborts its signal; advancing the clock past lease.endWaitMs afterwards, and the participant then resolving, logs no ceiling warning and starts no reclaim", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + + harness.engine.dispose(); + expect(participant.calls[0]?.signal.aborted).toBe(true); + harness.clock.advance(END_WAIT_MS + 1); + participant.calls[0]?.resolve(); + await flush(); + + expect(warningsFor(harness, grant.lease.id)).toEqual([]); + expect(reclaims(harness.driver)).toEqual([]); + }); + + it("with no participant, the reclaim starts without the clock advancing and the device record carries no deferred-reclaim marker", async () => { + const world = createWorld({ latencyMs: { reclaim: 20 } }); + const harness = await startEngine({ world }); + const grant = await harness.engine.request(request, asking("agent")); + + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + + expect(reclaims(harness.driver)).toHaveLength(1); + expect(stateOf(harness, grant.device.id)).toBe("reclaiming"); + expect(markerOf(harness, grant.device.id)).toBeUndefined(); + }); + + it("with a participant, the device is reclaiming with its deferred-reclaim marker while the participant runs, and the marker is gone once the device is ready", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + expect(stateOf(harness, grant.device.id)).toBe("reclaiming"); + expect(markerOf(harness, grant.device.id)).toBe(grant.lease.id); + + participant.calls[0]?.resolve(); + await flush(); + + expect(stateOf(harness, grant.device.id)).toBe("ready"); + expect(markerOf(harness, grant.device.id)).toBeUndefined(); + }); + + it("with running capacity full, while a participant waits, a lease request for another spec queues, and a noWait request gets NO_CAPACITY; the queued one is granted once the wait and the wipe end", async () => { + const { harness, participant } = await leaseAndRelease({ + ios: { maxDevices: 2, maxRunning: 1 }, + }); + expect(participant.calls).toHaveLength(1); + + const queued = harness.engine.request(otherRequest, asking("queued")); + const refused = harness.engine.request(otherRequest, { ...asking("hurried"), noWait: true }); + + await expect(refused).rejects.toBeInstanceOf(NoCapacityError); + await expect(settledOrPending(queued)).resolves.toBe("still pending"); + participant.calls[0]?.resolve(); + await flush(); + await harness.engine.settle(); + await expect(settledOrPending(queued)).resolves.toMatchObject({ + lease: { requesterId: "queued" }, + }); + }); + + it("a daemon stopped while a participant has not returned: a new engine on the same registry runs the full wipe, the device becomes ready, and no participant registered in the new engine is called", async () => { + const { grant, harness, participant } = await leaseAndRelease(); + expect(participant.calls).toHaveLength(1); + const replacement = new TestParticipant(); + + const restarted = await startEngine({ participants: [replacement], world: harness.world }); + await restarted.engine.convergeRunningCapacity(); + await flush(); + await restarted.engine.settle(); + + expect(reclaims(harness.driver)).toMatchObject([ + { arguments: [expect.anything(), { clean: "standard" }] }, + ]); + expect(stateOf(restarted, grant.device.id)).toBe("ready"); + expect(replacement.calls).toEqual([]); + }); + + it("a daemon stopped during the wipe that follows the wait also gets a full wipe at the next start", async () => { + const world = createWorld({ latencyMs: { reclaim: 20 } }); + const participant = new TestParticipant(); + const harness = await startEngine({ participants: [participant], world }); + const grant = await harness.engine.request(request, asking("agent")); + await harness.engine.release(grant.lease.id, "explicit"); + await flush(); + expect(participant.calls).toHaveLength(1); + participant.calls[0]?.resolve(); + await flush(); + expect(reclaims(world.driver)).toHaveLength(1); + const wipesBefore = reclaims(world.driver).length; + + const restarted = await startEngine({ world }); + await restarted.engine.convergeRunningCapacity(); + await flush(); + + expect(reclaims(world.driver).length - wipesBefore).toBe(1); + expect(reclaims(world.driver).at(-1)).toMatchObject({ + arguments: [expect.anything(), { clean: "standard" }], + }); + }); + + it("a release with a participant emits the same events in the same order as one without (lease.released, then device.reclaimed)", async () => { + const plain = await startEngine(); + const plainGrant = await plain.engine.request(request, asking("agent")); + const plainBefore = eventNames(plain).length; + await plain.engine.release(plainGrant.lease.id, "explicit"); + await flush(); + await plain.engine.settle(); + + const participant = new TestParticipant(); + const waiting = await startEngine({ participants: [participant] }); + const waitingGrant = await waiting.engine.request(request, asking("agent")); + const waitingBefore = eventNames(waiting).length; + await waiting.engine.release(waitingGrant.lease.id, "explicit"); + await flush(); + expect(participant.calls).toHaveLength(1); + participant.calls[0]?.resolve(); + await flush(); + await waiting.engine.settle(); + + const expected = eventNames(plain).slice(plainBefore); + expect(expected.indexOf("lease.released")).toBeLessThan(expected.indexOf("device.reclaimed")); + expect(eventNames(waiting).slice(waitingBefore)).toEqual(expected); + }); +}); diff --git a/src/leasing/lease-health-monitor.test.ts b/src/leasing/lease-health-monitor.test.ts index 4f0ca278..34140b81 100644 --- a/src/leasing/lease-health-monitor.test.ts +++ b/src/leasing/lease-health-monitor.test.ts @@ -63,6 +63,7 @@ function config(overrides: Partial = {}): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/leasing/lease-ports.ts b/src/leasing/lease-ports.ts index ecf44d78..b67b3b56 100644 --- a/src/leasing/lease-ports.ts +++ b/src/leasing/lease-ports.ts @@ -1,4 +1,10 @@ -import type { DeviceRequest, DeviceSpec, LeaseGrant, LeaseRecord } from "../core/index.js"; +import type { + DeviceRequest, + DeviceSpec, + LeaseGrant, + LeaseRecord, + Platform, +} from "../core/index.js"; import type { WaitingRequest } from "./lease-request-book.js"; import type { LeaseReleaseReason } from "./lease-release-coordinator.js"; import type { LeaseRequestOptions } from "./wait-queue.js"; @@ -20,6 +26,18 @@ export interface QueueControl { waitingRequests(): readonly WaitingRequest[]; } +export interface EndedLease { + readonly leaseId: string; + readonly ownerId: string; + readonly deviceId: string; + readonly platform: Platform; + readonly reason: "explicit" | "killed" | "expired" | "device-lost"; +} + +export interface LeaseEndParticipant { + leaseEnded(ended: EndedLease, signal: AbortSignal): Promise; +} + export interface DeviceModeReader { servesDefaultMode(spec: DeviceSpec): boolean; } diff --git a/src/leasing/lease-release-coordinator.test.ts b/src/leasing/lease-release-coordinator.test.ts index c5df9591..3edaadc3 100644 --- a/src/leasing/lease-release-coordinator.test.ts +++ b/src/leasing/lease-release-coordinator.test.ts @@ -660,6 +660,47 @@ describe("LeaseReleaseCoordinator", () => { expect(harness.registry.snapshot.leases.map((lease) => lease.id)).toEqual([held.lease.id]); }); + it("endForAbsentDevice calls the lease-end participant after the decision section has ended, not inside it", async () => { + const harness = await createHarness(); + const held = await grantPhysical(harness, "UDID-A"); + let inSection = false; + let calledInSection: boolean | undefined; + const coordinator = new LeaseReleaseCoordinator({ + claims: harness.claims, + decisions: { + run: async (operation) => { + inSection = true; + try { + return await harness.decisions.run(operation); + } finally { + inSection = false; + } + }, + }, + leaseEnd: { + clock: harness.clock, + endWaitMs: 1_000, + participants: [ + { + leaseEnded: async () => { + calledInSection = inSection; + }, + }, + ], + }, + lifecycle: harness.lifecycle, + notifyAvailability: () => undefined, + reclaim: harness.reclaim, + registry: harness.registry, + }); + + await expect(coordinator.endForAbsentDevice(held.lease.id, "not-present")).resolves.toBe( + true, + ); + + expect(calledInSection).toBe(false); + }); + it("endForAbsentDevice answers false when no lease exists at all", async () => { const harness = await createHarness(); diff --git a/src/leasing/lease-release-coordinator.ts b/src/leasing/lease-release-coordinator.ts index c119a6a1..64de0f4d 100644 --- a/src/leasing/lease-release-coordinator.ts +++ b/src/leasing/lease-release-coordinator.ts @@ -1,4 +1,4 @@ -import { type Logger, NoopLogger } from "../ports/index.js"; +import { type Clock, type Logger, NoopLogger, type TimerHandle } from "../ports/index.js"; import { type AbsentReason, type DeviceOperationClaims, @@ -7,6 +7,7 @@ import { type SerializedDecision, type ReclaimCoordinator, } from "../core/index.js"; +import type { EndedLease, LeaseEndParticipant } from "./lease-ports.js"; export type LeaseReleaseReason = "explicit" | "killed" | "device-lost"; @@ -53,6 +54,12 @@ export interface LeaseReleaseRegistry { } export interface LeaseReleaseCoordinatorOptions { + /** why: absent when no participant is registered, so that path runs exactly as before. */ + readonly leaseEnd?: { + readonly clock: Clock; + readonly endWaitMs: number; + readonly participants: readonly LeaseEndParticipant[]; + }; readonly claims: Pick; readonly decisions: Pick; readonly lifecycle: LeaseReleaseLifecycle; @@ -64,6 +71,11 @@ export interface LeaseReleaseCoordinatorOptions { type ReclaimMode = "await" | "background" | "none"; +interface ParticipantCalls { + readonly done: Promise; + readonly cutShort: () => void; +} + /** * why: lease commits happen in the serialized decision section; reclaiming is deliberately outside it. * Concurrent `releaseAll` calls are not idempotent: overlapping snapshots can yield UnknownLeaseError. @@ -75,7 +87,10 @@ export class LeaseReleaseCoordinator readonly #backgroundReclaims = new Set>(); readonly #logger: Logger; readonly #maintenanceWaiters: (() => void)[] = []; + readonly #participantCalls = new Set(); + #disposed = false; #maintenanceDepth = 0; + #stopping = false; constructor(private readonly options: LeaseReleaseCoordinatorOptions) { this.#logger = options.logger?.child("lease-release-coordinator") ?? new NoopLogger(); @@ -119,8 +134,12 @@ export class LeaseReleaseCoordinator await this.#runNormal(async () => { switch (ending.device) { case "missing": - await this.options.decisions.run(() => - this.options.lifecycle.endForMissingDevice(leaseId, ending.reason), + this.#callParticipants( + await this.options.decisions.run(() => + this.options.lifecycle.endForMissingDevice(leaseId, ending.reason), + ), + ending.reason, + false, ); return; case "reclaim": @@ -142,18 +161,20 @@ export class LeaseReleaseCoordinator absentReason: AbsentReason, proceed: () => boolean = () => true, ): Promise { - return this.#runNormal(() => - this.options.decisions.run(async () => { + return this.#runNormal(async () => { + const ended = await this.options.decisions.run(async () => { if ( !proceed() || !this.options.registry.snapshot.leases.some((lease) => lease.id === leaseId) ) { - return false; + return undefined; } - await this.options.lifecycle.endForAbsentDevice(leaseId, absentReason); - return true; - }), - ); + return this.options.lifecycle.endForAbsentDevice(leaseId, absentReason); + }); + if (ended === undefined) return false; + this.#callParticipants(ended, "device-lost", false); + return true; + }); } /** why: under the maintenance admission every release takes, so no watcher write lands while an operator reset walks the devices. */ @@ -179,12 +200,21 @@ export class LeaseReleaseCoordinator }); } + /** why: a graceful stop waits only for wipes; the participants' wait is cut short and later endings do not wait. */ async settleBackgroundReclaims(): Promise { + this.#stopping = true; + this.#cutShortWaits(); while (this.#backgroundReclaims.size > 0) { await Promise.allSettled(this.#backgroundReclaims); } } + /** why: a disposed engine runs no more driver work; a participant that returns late starts no reclaim. */ + dispose(): void { + this.#disposed = true; + this.#cutShortWaits(); + } + async renew(leaseId: string, ttlMs?: number): Promise { return this.#runNormal(() => this.options.decisions.run(() => this.options.lifecycle.renew(leaseId, ttlMs)), @@ -200,6 +230,7 @@ export class LeaseReleaseCoordinator // a device still `reclaiming` is neither `ready` nor `shutdown`, so an operator // reset would silently skip it (`NukeService#canOperate`) and leave a device // running that the reset was supposed to take down. + this.#cutShortWaits(); while (this.#activeWorkflows.size > 0 || this.#backgroundReclaims.size > 0) { await Promise.allSettled([...this.#activeWorkflows, ...this.#backgroundReclaims]); } @@ -243,10 +274,16 @@ export class LeaseReleaseCoordinator return this.options.lifecycle.beginRelease( leaseId, reason, - options.reclaim === "none" ? { deferReclaim: true } : {}, + options.reclaim === "none" || (options.reclaim === "background" && this.#hasParticipants) + ? { deferReclaim: true } + : {}, ); }); - if (released === undefined || options.reclaim === "none") return; + if (released === undefined) return; + if (options.reclaim === "none") { + this.#callParticipants(released, reason, false); + return; + } if (options.reclaim === "background") { // The registry-only half committed above is the whole of what a releasing // caller needs: the lease record is gone, `lease.released` has been emitted, @@ -265,36 +302,153 @@ export class LeaseReleaseCoordinator // immediately (not one-after-another): queuing would let a healthy reclaim sit // idle in `reclaiming` waiting its turn, which is exactly the state age a // stalled-transition detector would misread as a stall. - this.#reclaimInBackground(released); + this.#reclaimInBackground(released, reason); return; } + // why: under maintenance the call gets an aborted signal and nothing waits for it, so a + // participant that ignores the signal cannot hang `nuke`. + this.#callParticipants(released, reason, false); await this.options.reclaim.reclaim(released); } /** why: the claim lasts as long as the reclaim, so startup recovery and doctor's stalled-transition check take a live reclaim for neither orphaned nor stalled. */ - #reclaimInBackground(released: ReleasedLease): void { + #reclaimInBackground(released: ReleasedLease, reason: EndedLease["reason"]): void { const claim = this.options.claims.tryClaim(released.device.id, "reclaim"); - const reclaim = this.options.reclaim - .reclaim(released) - .catch((error: unknown) => { - this.#logger.error("background reclaim failed", { - deviceId: released.device.id, - error: error instanceof Error ? error.message : String(error), - leaseId: released.lease.id, + const calls = this.#callParticipants(released, reason, true); + const run = (): Promise => + this.options.reclaim + .reclaim(released) + .catch((error: unknown) => { + this.#logger.error("background reclaim failed", { + deviceId: released.device.id, + error: error instanceof Error ? error.message : String(error), + leaseId: released.lease.id, + }); + }) + .finally(() => { + claim?.release(); + // why: `ReclaimCoordinator` already wakes the queue on every path that frees a + // device, and it does so before this claim is released, so the wake-up here is a + // safety net, not the one that serves a waiter: it runs after the claim is gone, + // so matching sees the device even if an earlier wake-up raced the release. + this.options.notifyAvailability(); }); - }) - .finally(() => { - claim?.release(); - // `ReclaimCoordinator` already wakes the queue on every path that frees a - // device, and it does so before this claim is released, so the wake-up here is a - // safety net, not the one that serves a waiter: it runs after the claim is gone, - // so matching sees the device even if an earlier wake-up raced the release. - this.options.notifyAvailability(); - }); + // why: the reclaim's own promise chain is not wrapped, so the claim is released in as few + // steps as without a participant; the queue's wake-up races that release. + const reclaim = + calls === undefined + ? run() + : new Promise((resolve) => { + void calls.done.then(() => { + if (this.#disposed) { + claim?.release(); + resolve(); + return; + } + void run().finally(resolve); + }); + }); this.#backgroundReclaims.add(reclaim); void reclaim.finally(() => this.#backgroundReclaims.delete(reclaim)); } + get #hasParticipants(): boolean { + return this.options.leaseEnd !== undefined; + } + + #cutShortWaits(): void { + for (const calls of this.#participantCalls) calls.cutShort(); + } + + /** + * why: participants are called after the commit and outside the serialized decision section: one + * may take minutes. One budget per ended lease covers all of them and never restarts. Leasing + * attaches its handlers here, so a rejection after it stopped waiting is logged, never unhandled. + * Returns calls to wait for, or `undefined` when there is nothing to wait for. + */ + #callParticipants( + released: ReleasedLease, + reason: EndedLease["reason"], + waitable: boolean, + ): ParticipantCalls | undefined { + const { leaseEnd } = this.options; + if (leaseEnd === undefined) return undefined; + const { clock, endWaitMs, participants } = leaseEnd; + const ended: EndedLease = { + deviceId: released.device.id, + leaseId: released.lease.id, + ownerId: released.lease.ownerId, + platform: released.device.spec.platform, + reason, + }; + const context = { deviceId: ended.deviceId, leaseId: ended.leaseId }; + const controller = new AbortController(); + const aborted = this.#maintenanceDepth > 0 || this.#stopping || this.#disposed; + if (aborted) controller.abort(); + const settled = participants.map(async (participant) => + this.#callOne(participant, ended, controller.signal), + ); + // why: aborted before the call: nothing waits and no ceiling is armed. + if (aborted) return undefined; + const calls = this.#armCeiling(clock, endWaitMs, controller, context, settled); + return waitable ? calls : undefined; + } + + async #callOne( + participant: LeaseEndParticipant, + ended: EndedLease, + signal: AbortSignal, + ): Promise { + try { + await participant.leaseEnded(ended, signal); + } catch (error: unknown) { + this.#logger.error("lease-end participant failed", { + deviceId: ended.deviceId, + error: error instanceof Error ? error.message : String(error), + leaseId: ended.leaseId, + }); + } + } + + #armCeiling( + clock: Clock, + endWaitMs: number, + controller: AbortController, + context: { readonly deviceId: string; readonly leaseId: string }, + settled: readonly Promise[], + ): ParticipantCalls { + let timer: TimerHandle | undefined; + let finish!: () => void; + const done = new Promise((resolve) => { + finish = resolve; + }); + const calls: ParticipantCalls = { + cutShort: () => { + stop(); + controller.abort(); + }, + done, + }; + const stop = (): void => { + if (timer !== undefined) clock.cancel(timer); + timer = undefined; + this.#participantCalls.delete(calls); + finish(); + }; + this.#participantCalls.add(calls); + timer = clock.setTimer(endWaitMs, () => { + timer = undefined; + this.#logger.warn("lease-end participants passed the wait ceiling", { + ...context, + endWaitMs, + }); + controller.abort(); + stop(); + }); + void Promise.all(settled).then(stop); + return calls; + } + async #releaseAll( reason: "explicit" | "killed", reclaim: ReclaimMode, diff --git a/src/leasing/physical-device-watcher.test.ts b/src/leasing/physical-device-watcher.test.ts index 9ec0e098..d38c426b 100644 --- a/src/leasing/physical-device-watcher.test.ts +++ b/src/leasing/physical-device-watcher.test.ts @@ -81,6 +81,7 @@ function config(options: HarnessOptions): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, }, capacity: { strategy: "resource", diff --git a/src/leasing/physical-leasing.test.ts b/src/leasing/physical-leasing.test.ts index 1b683d38..f6525238 100644 --- a/src/leasing/physical-leasing.test.ts +++ b/src/leasing/physical-leasing.test.ts @@ -75,6 +75,7 @@ function config(leaseOverrides: object = {}): Config { requestRetentionMs: 600_000, maxRequestRecords: 10_000, maxPerRequester: 1, + endWaitMs: 600_000, ...leaseOverrides, }, capacity: { diff --git a/src/leasing/testing.ts b/src/leasing/testing.ts index 15a68c53..3c1d5aa8 100644 --- a/src/leasing/testing.ts +++ b/src/leasing/testing.ts @@ -21,7 +21,12 @@ import { type SerializedDecision, } from "../core/index.js"; import { createLeasing, type Leasing } from "./create-leasing.js"; -import type { DeviceModeReader, LeaseCommands, QueueControl } from "./lease-ports.js"; +import type { + DeviceModeReader, + LeaseCommands, + LeaseEndParticipant, + QueueControl, +} from "./lease-ports.js"; import type { LeaseHealthMonitor } from "./lease-health-monitor.js"; /** @@ -43,6 +48,7 @@ export interface TestEngineOptions { readonly describeFailure?: ((error: unknown) => LeaseRequestFailure) | undefined; readonly defaultModes?: Readonly>> | undefined; readonly modelPreferences?: ModelPreferences | undefined; + readonly leaseEndParticipants?: readonly LeaseEndParticipant[] | undefined; } /** @@ -96,6 +102,7 @@ export function createTestEngine(options: TestEngineOptions): TestEngine { describeFailure: options.describeFailure, eventBus: options.eventBus, idGenerator: options.idGenerator, + leaseEndParticipants: options.leaseEndParticipants, logger: options.logger, modelPreferences: options.modelPreferences, });