diff --git a/apps/server/src/scheduling/DueHint.test.ts b/apps/server/src/scheduling/DueHint.test.ts new file mode 100644 index 0000000000000..d8e7b4fdbc879 --- /dev/null +++ b/apps/server/src/scheduling/DueHint.test.ts @@ -0,0 +1,26 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import * as DueHint from "./DueHint.ts"; + +const NOW = Date.parse("2026-11-14T12:00:00.000Z"); +const HOUR = 3_600_000; + +it.effect("a recorded hint keeps the poller idle until its work is due", () => + Effect.gen(function* () { + const due = DueHint.make(); + yield* due.record(yield* due.version, NOW + 30_000, NOW); + assert.isTrue(yield* due.isIdle(NOW + 5_000)); + assert.isFalse(yield* due.isIdle(NOW + 30_000)); + }), +); + +it.effect("a hint recorded before the wall clock was set back does not keep the poller idle", () => + Effect.gen(function* () { + const due = DueHint.make(); + // The next poll has work in 30 s. + yield* due.record(yield* due.version, NOW + 30_000, NOW); + // The wall clock is set back an hour; a scheduler tick 5 s of real time later. + assert.isFalse(yield* due.isIdle(NOW - HOUR + 5_000)); + }), +); diff --git a/apps/server/src/scheduling/DueHint.ts b/apps/server/src/scheduling/DueHint.ts index 2df8af5f0c536..e981ef225afc9 100644 --- a/apps/server/src/scheduling/DueHint.ts +++ b/apps/server/src/scheduling/DueHint.ts @@ -17,10 +17,14 @@ export const make = () => { hint = null; }), version: Effect.sync(() => version), + // A wall clock set back (now before the check) is not idle: the hint's times are stale. isIdle: (nowMs: number) => Effect.sync( () => - hint !== null && hint.dueAtMs > nowMs && nowMs - hint.checkedAtMs < DUE_HINT_MAX_AGE_MS, + hint !== null && + hint.dueAtMs > nowMs && + nowMs >= hint.checkedAtMs && + nowMs - hint.checkedAtMs < DUE_HINT_MAX_AGE_MS, ), record: (readVersion: number, dueAtMs: number, nowMs: number) => Effect.sync(() => { diff --git a/apps/server/src/slots/SlotGate.test.ts b/apps/server/src/slots/SlotGate.test.ts index ea037a2154205..a022892194d4d 100644 --- a/apps/server/src/slots/SlotGate.test.ts +++ b/apps/server/src/slots/SlotGate.test.ts @@ -999,6 +999,88 @@ describe("SlotGate", () => { }), ); + // While a re-ask on another account (or none) is not kept, the old hold still counts + // against its own account, so a dropped re-ask never leaves that account booked twice. + it.effect( + "a re-ask not kept yet still counts the old hold against its account, once on the machine", + () => + Effect.gen(function* () { + const world = newWorld(); + const third = ThreadId.make("thread:slot-third"); + yield* run( + world, + waitsOn({ maxPerMachine: 2, maxPerAccount: { [SECOND]: 1 }, reserveSeconds: 60 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + assert.deepStrictEqual( + types(yield* grant({ key: "h0", threadId: holder, account: SECOND })), + ["granted"], + ); + yield* gate.keep(["h0"]); + // One sweep: the holder re-asks with no account, then `other` asks for SECOND. + // The holder's SECOND hold is hidden but still counts there: `other` waits. + assert.deepStrictEqual( + types( + yield* grant( + { key: "w1", threadId: holder }, + { key: "w2", threadId: other, account: SECOND }, + ), + ), + ["granted", "full"], + ); + // The settings screen shows the hidden hold on SECOND too (1 of 1, not 0 of 1). + assert.deepStrictEqual( + (yield* gate.usage)?.accounts.find((account) => account.id === SECOND), + { id: SECOND, used: 1, max: 1 }, + ); + assert.strictEqual((yield* acquireAs(other, SECOND)).type, "refuse"); + // Not twice against the machine: the holder counts once (its new hold), so a + // third thread still gets the second machine slot. + assert.deepStrictEqual(types(yield* grant({ key: "w3", threadId: third })), [ + "granted", + ]); + // The re-ask is cancelled before it is written: the old hold is back, and SECOND + // holds it once (not over its limit), for the holder only. + yield* gate.drop(["w1"]); + assert.deepStrictEqual( + types(yield* grant({ key: "w2", threadId: other, account: SECOND })), + ["full"], + ); + const usage = yield* gate.usage; + assert.deepStrictEqual( + usage?.accounts.find((account) => account.id === SECOND), + { id: SECOND, used: 1, max: 1 }, + ); + assert.strictEqual((yield* acquireAs(holder, SECOND)).type, "allow"); + }), + ); + // Kept this time: the old SECOND hold is used up, and `other` gets SECOND next sweep. + yield* run( + newWorld(), + waitsOn({ maxPerMachine: 3, maxPerAccount: { [SECOND]: 1 }, reserveSeconds: 60 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + yield* grant({ key: "h0", threadId: holder, account: SECOND }); + yield* gate.keep(["h0"]); + assert.deepStrictEqual( + types( + yield* grant( + { key: "w1", threadId: holder }, + { key: "w2", threadId: other, account: SECOND }, + ), + ), + ["granted", "full"], + ); + yield* gate.keep(["w1"]); + assert.deepStrictEqual( + types(yield* grant({ key: "w2", threadId: other, account: SECOND })), + ["granted"], + ); + }), + ); + }), + ); + it.effect("a thread already holding a slot that is refused keeps the slot it holds", () => Effect.gen(function* () { const world = newWorld(); diff --git a/apps/server/src/slots/SlotGate.ts b/apps/server/src/slots/SlotGate.ts index 6f8ad7a028e73..f9d575c755c29 100644 --- a/apps/server/src/slots/SlotGate.ts +++ b/apps/server/src/slots/SlotGate.ts @@ -6,6 +6,7 @@ */ import type { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import { + accountOnlyThreads, effectiveSlotLimits, limitsCount, liveReservations, @@ -140,8 +141,9 @@ const make = Effect.gen(function* () { /** Keys used up, until their end: the store's copy stops counting even if clearing it failed. */ const usedKeys = new Map(); /** - * A re-ask's grant not kept yet, by its key: the thread's earlier holds, hidden (they do not - * count) but not used up, so a grant that is dropped gives them back. + * A re-ask's grant not kept yet, by its key: the thread's earlier holds, hidden (they count only + * against their own account, see `hiddenHeld`) but not used up, so a grant that is dropped gives + * them back. */ const replacing = new Map>(); const hiddenKeys = () => @@ -209,6 +211,18 @@ const make = Effect.gen(function* () { ); }); + /** + * The live holds that re-asks not kept yet hide, of threads other than `threadId`. They still + * count against their own account (never the machine, where the new hold counts), so a slot + * that a dropped re-ask gives back is never booked twice. None while waits are off. + */ + const hiddenHeld = (waitsOn: boolean, nowMs: number, threadId: string) => + waitsOn + ? liveReservations([...replacing.values()].flat(), nowMs).filter( + (old) => old.threadId !== threadId && !usedKeys.has(old.key), + ) + : []; + const freeMemoryBytes = (limits: SlotLimits) => limits.minFreeMemoryBytes === null ? Effect.succeed(null) @@ -294,6 +308,7 @@ const make = Effect.gen(function* () { reservations: onePerThread( reservations.filter((reservation) => reservation.threadId !== input.callerThreadId), ), + accountOnly: hiddenHeld(waitsOn, nowMs, input.callerThreadId), targetInstanceId: input.targetInstanceId, freeMemoryBytes: yield* freeMemoryBytes(limits), }); @@ -374,6 +389,8 @@ const make = Effect.gen(function* () { reservations: onePerThread( reservations.filter((reservation) => reservation.threadId !== request.threadId), ), + // Read again for each request: a grant earlier in this call may have hidden holds. + accountOnly: hiddenHeld(settings.value.waitFor.enabled, nowMs, request.threadId), targetInstanceId: request.account ?? "", freeMemoryBytes: memoryBytes, }); @@ -446,6 +463,8 @@ const make = Effect.gen(function* () { ? [] : liveStarting(yield* Ref.get(entries), read.visible, nowMs, limits.startGraceMs); const reservations = onePerThread(yield* liveHeld(settings.value.waitFor.enabled, nowMs)); + // Holds hidden by a re-ask not kept yet: on their own account only, as `decide` counts them. + const hidden = hiddenHeld(settings.value.waitFor.enabled, nowMs, ""); const accountMax = (id: string) => { const max = agentSlots.maxPerAccount[id as ProviderInstanceId]; return max === undefined || max <= 0 ? null : max; @@ -454,7 +473,7 @@ const make = Effect.gen(function* () { ...Object.keys(agentSlots.maxPerAccount).filter((id) => accountMax(id) !== null), ...read.running.map((agent) => agent.instanceId), ...starting.map((entry) => entry.instanceId), - ...reservations.flatMap((reservation) => + ...[...reservations, ...hidden].flatMap((reservation) => reservation.account === undefined ? [] : [reservation.account], ), ]); @@ -469,7 +488,8 @@ const make = Effect.gen(function* () { used: read.running.filter((agent) => agent.instanceId === id).length + starting.filter((entry) => entry.instanceId === id).length + - reservations.filter((reservation) => reservation.account === id).length, + reservations.filter((reservation) => reservation.account === id).length + + accountOnlyThreads(reservations, hidden, id), max: accountMax(id), })), freeMemoryMb: Math.floor(memory.availableMemoryBytes / (1024 * 1024)), diff --git a/apps/server/src/waits/WaitPrJudge.test.ts b/apps/server/src/waits/WaitPrJudge.test.ts index 65d2102f7be57..058f61a86e1ac 100644 --- a/apps/server/src/waits/WaitPrJudge.test.ts +++ b/apps/server/src/waits/WaitPrJudge.test.ts @@ -15,7 +15,9 @@ import { } from "@t3tools/contracts"; import type { WaitForInput, WaitRecord } from "@t3tools/contracts/waitFor"; import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -32,6 +34,7 @@ import type { WaitDelivery } from "./WaitDelivery.ts"; import type { WaitJudge } from "./WaitJudges.ts"; import { makePrJudge } from "./WaitPrJudge.ts"; import * as WaitService from "./WaitService.ts"; +import { memoryFileSystem } from "./WaitService.testkit.ts"; import { WAIT_STORE_FILE } from "./WaitStore.ts"; const NOW = Date.parse("2026-11-15T12:00:00.000Z"); @@ -771,6 +774,87 @@ it.layer(NodeServices.layer)("WaitPrJudge", (it) => { }, ); + it.effect( + "a pin made while the new wait is being written is saved with it, so a changed remote is never read (#118)", + () => { + const host = fakeHost(); + return Effect.gen(function* () { + yield* TestClock.setTime(NOW); + host.prs.set("acme/widgets#7", { state: "open", headSha: HEAD_A }); + // In memory, so every write answers at once while the test clock moves. + const filePath = `/waits/${WAIT_STORE_FILE}`; + const memoryFs = yield* FileSystem.FileSystem; + const writing = yield* Deferred.make(); + const pinned = yield* Deferred.make(); + let armed = true; + // The wait's first write ends in a rename onto the store; it is held there until the pin. + const fs: FileSystem.FileSystem = { + ...memoryFs, + rename: (from, to) => + armed && to === filePath + ? Effect.gen(function* () { + armed = false; + yield* Deferred.succeed(writing, undefined); + yield* Deferred.await(pinned); + return yield* memoryFs.rename(from, to); + }) + : memoryFs.rename(from, to), + }; + const letFibersRun = Effect.gen(function* () { + for (let turn = 0; turn < 20; turn += 1) yield* Effect.yieldNow; + }); + const prJudge = yield* makePrJudge(); + // At registration the read answers only after the 3 s limit, while the wait is written. + const lateRead: WaitJudge<"pr"> = (target, context) => + context.waitId !== null + ? prJudge(target, context) + : Deferred.await(writing).pipe( + Effect.andThen( + prJudge(target, { + ...context, + pin: (next) => + context.pin!(next).pipe(Effect.andThen(Deferred.succeed(pinned, undefined))), + }), + ), + ); + const service = yield* makeService(filePath, lateRead).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + ); + const registering = yield* Effect.forkChild( + register(service, prWait({ number: 7, until: "merged" })), + ); + // One second at a time, so each forked step has set its timer before the clock moves. + for (let second = 0; second < 5; second += 1) { + yield* letFibersRun; + yield* TestClock.adjust(1_000); + } + assert.strictEqual((yield* Fiber.join(registering)).state, "waiting"); + const pinnedSpec = { + number: 7, + repository: "acme/widgets", + host: "github.com", + until: "merged" as const, + }; + assert.deepStrictEqual((yield* storedWaits(filePath))[0]?.spec, pinnedSpec); + + // The project's remote flips before the next check: the wait still reads acme/widgets. + host.projects.set(PROJECT, shell(PROJECT, "upstream/widgets")); + host.remotes.set(`/work/${PROJECT}`, identityOf("upstream/widgets")); + host.prs.set("upstream/widgets#7", { state: "merged", headSha: HEAD_B }); + for (let pass = 0; pass < 2; pass += 1) { + yield* TestClock.adjust(SWEEP); + yield* service.judgeDue(); + } + assert.isFalse(host.reads.some((read) => read.includes("upstream/"))); + assert.deepStrictEqual((yield* storedWaits(filePath))[0]?.spec, pinnedSpec); + assert.strictEqual((yield* storedWaits(filePath))[0]?.state, "waiting"); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(Layer.merge(host.layer, ServerSettingsService.layerTest(ON))), + ); + }, + ); + it.effect("a pin never overwrites a wait cancelled while it was read", () => { const host = fakeHost(); return Effect.gen(function* () { diff --git a/apps/server/src/waits/WaitService.test.ts b/apps/server/src/waits/WaitService.test.ts index ef797676aaa9e..7bc73688cf049 100644 --- a/apps/server/src/waits/WaitService.test.ts +++ b/apps/server/src/waits/WaitService.test.ts @@ -3,6 +3,7 @@ import { assert, it } from "@effect/vitest"; import { ProjectId, ThreadId } from "@t3tools/contracts"; import type { WaitForInput, WaitRecord } from "@t3tools/contracts/waitFor"; import { type Judgement, MAX_OPEN_WAITS } from "@t3tools/shared/waitFor"; +import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -2542,3 +2543,457 @@ it.layer(NodeServices.layer)("WaitService: carry from #90 (fix round 2 on #91)", }).pipe(inMemory, Effect.provide(ServerSettingsService.layerTest(ON))), ); }); + +it.layer(NodeServices.layer)("WaitService: monotonic clock and early messages (#95)", (it) => { + const ALPHABET = "abcdefghijklmnopqrstuvwxyz234567"; + const idOf = (n: number) => + `w_${Array.from({ length: 10 }, (_, index) => ALPHABET[Math.floor(n / 32 ** (9 - index)) % 32]).join("")}`; + // In memory (see `memoryFileSystem`), so every write answers at once while the test clock moves. + const filePath = `/waits/state/${WAIT_STORE_FILE}`; + const watched = (name: string) => `/waits/${name}`; + const letFibersRun = Effect.gen(function* () { + for (let turn = 0; turn < 20; turn += 1) yield* Effect.yieldNow; + }); + const fileRecord = (n: number, path: string) => + record({ id: idOf(n), key: n.toString(16).padStart(16, "0"), spec: { path, until: "exists" } }); + + /** + * The test clock, with its wall time and its monotonic time moved by their own `shift`: a time + * sync or a restored snapshot moves the wall clock only (timers keep running, as Node's are + * monotonic), and a new process starts its monotonic clock again. + */ + const shiftedClock = Effect.gen(function* () { + const test = yield* TestClock.testClockWith(Effect.succeed); + const shift = { wall: 0, monotonic: 0 }; + const wall = () => test.currentTimeMillisUnsafe() + shift.wall; + const monotonic = () => test.monotonicTimeNanosUnsafe() + BigInt(shift.monotonic) * 1_000_000n; + const clock = { + ...test, + currentTimeMillisUnsafe: wall, + currentTimeMillis: Effect.sync(wall), + currentTimeNanosUnsafe: () => BigInt(wall()) * 1_000_000n, + currentTimeNanos: Effect.sync(() => BigInt(wall()) * 1_000_000n), + monotonicTimeNanosUnsafe: monotonic, + monotonicTimeNanos: Effect.sync(monotonic), + } as unknown as Clock.Clock; + return { clock, shift }; + }); + + /** + * Three stored file waits: `dead/a` and `dead/b` hang in the server (a dead mount), `ready` + * exists. A file check in its own process answers, and is counted. + */ + const deadMount = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + yield* fs.writeFileString(watched("ready"), ""); + yield* seed(filePath, [ + fileRecord(1, watched("dead/a")), + fileRecord(2, watched("dead/b")), + fileRecord(3, watched("ready")), + ]); + const exists = (path: string) => + fs.exists(path).pipe( + Effect.orDie, + Effect.map((found): Judgement => + found ? { type: "met", outcome: "exists", detail: "It exists." } : { type: "notYet" }, + ), + ); + const childChecks: string[] = []; + const service = yield* makeService({ + filePath, + judges: { + file: (target) => + target.spec.path.includes("/dead/") + ? (Effect.never as Effect.Effect) + : exists(target.spec.path), + }, + fileChildJudge: (target) => + Effect.sync(() => childChecks.push(target.spec.path)).pipe( + Effect.andThen(exists(target.spec.path)), + ), + }); + const ready = Effect.map( + storedWaits(filePath), + (waits) => waits.find((wait) => wait.id === idOf(3))?.state, + ); + return { service, childChecks, ready }; + }); + + /** One pass, with 15 s of the test clock for its checks. */ + const pass = (service: WaitService.WaitService) => + Effect.gen(function* () { + const running = yield* Effect.forkChild(service.judgeDue()); + yield* letFibersRun; + yield* stepSeconds(15); + yield* letFibersRun; + const done = running.pollUnsafe() !== undefined; + if (!done) yield* Fiber.interrupt(running); + return done; + }); + + it.effect( + "a met file wait's message goes out at once, while slow checks of another kind still run", + () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const fs = yield* FileSystem.FileSystem; + yield* fs.writeFileString(watched("ready"), ""); + yield* seed(filePath, [ + fileRecord(1, watched("ready")), + record({ + id: idOf(2), + key: "2".padStart(16, "0"), + kind: "pr", + spec: { number: 1, until: "checks" }, + }), + record({ + id: idOf(3), + key: "3".padStart(16, "0"), + kind: "pr", + spec: { number: 2, until: "checks" }, + }), + ]); + const sentAt: number[] = []; + const service = yield* makeService({ + filePath, + delivery: { + receiver: () => Effect.succeed({ type: "ok" }), + deliver: () => + Effect.map(Clock.currentTimeMillis, (now) => { + sentAt.push(now - NOW); + return { type: "delivered" } as const; + }), + }, + judges: { + file: () => + Effect.succeed({ type: "met", outcome: "exists", detail: "It exists." }), + // Healthy but slow (under their 60 s limit), like a slow pull request host. + pr: () => Effect.sleep(55_000).pipe(Effect.as({ type: "notYet" })), + }, + }); + const sweep = yield* Effect.forkChild(service.runDue()); + yield* letFibersRun; + yield* stepSeconds(5); + assert.strictEqual(sweep.pollUnsafe(), undefined, "the pull request checks still run"); + assert.deepStrictEqual(sentAt, [0]); + yield* stepSeconds(60); + yield* letFibersRun; + assert.isDefined(sweep.pollUnsafe(), "the sweep ended"); + // Sent once: the deliver step after the checks finds nothing more to send. + assert.deepStrictEqual(sentAt, [0]); + assert.strictEqual( + (yield* storedWaits(filePath)).find((wait) => wait.id === idOf(1))?.state, + "delivered", + ); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); + + it.effect( + "the wall clock set back an hour while two file checks hang: they still turn late, so the met wait is judged in its own process", + () => + Effect.gen(function* () { + const { clock, shift } = yield* shiftedClock; + yield* Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const { service, childChecks, ready } = yield* deadMount; + // a and b take both places and hang; ready waits for a place past the start window. + assert.isTrue(yield* pass(service)); + assert.strictEqual(yield* ready, "waiting"); + shift.wall = -HOUR; + yield* TestClock.adjust(5_000); + assert.isTrue(yield* pass(service)); + assert.deepStrictEqual(childChecks, [watched("ready")]); + assert.strictEqual(yield* ready, "delivering"); + }).pipe(Effect.provideService(Clock.Clock, clock)); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); + + it.effect("the wall clock set back an hour in the middle of a pass does not hold the pass", () => + Effect.gen(function* () { + const { clock, shift } = yield* shiftedClock; + yield* Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const { service } = yield* deadMount; + const running = yield* Effect.forkChild(service.judgeDue()); + yield* letFibersRun; + yield* stepSeconds(5); + shift.wall = -HOUR; + // Every timer of the pass is at most 12 s long. + yield* stepSeconds(30); + yield* letFibersRun; + const done = running.pollUnsafe() !== undefined; + if (!done) yield* Fiber.interrupt(running); + assert.isTrue(done, "the pass is still waiting after the clock moved back"); + }).pipe(Effect.provideService(Clock.Clock, clock)); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); + + /** Every stored time (`at`, or a key ending in At, Until or Since) is an ISO wall time near NOW. */ + const assertWallTimesOnly = (value: unknown, where = "store"): void => { + if (Array.isArray(value)) { + value.forEach((item, index) => assertWallTimesOnly(item, `${where}[${index}]`)); + return; + } + if (value === null || typeof value !== "object") return; + for (const [key, field] of Object.entries(value)) { + if (/^at$|(At|Until|Since)$/.test(key)) { + assert.strictEqual(typeof field, "string", `${where}.${key} is not an ISO wall time`); + const ms = Date.parse(field as string); + assert.isTrue( + Math.abs(ms - NOW) < 6 * HOUR, + `${where}.${key} = ${String(field)} is not a wall time near now`, + ); + } else assertWallTimesOnly(field, `${where}.${key}`); + } + }; + + it.effect( + "after a restart on the store the old run left (a new monotonic clock, a day behind): stuck checks still turn late and every stored time is a wall time", + () => + Effect.gen(function* () { + const { clock, shift } = yield* shiftedClock; + yield* Effect.gen(function* () { + yield* TestClock.setTime(NOW); + // The monotonic clock counts from the machine's boot, far from the wall clock. + shift.monotonic = -7 * 24 * HOUR; + const fs = yield* FileSystem.FileSystem; + yield* fs.writeFileString(watched("ready"), ""); + yield* seed(filePath, [ + fileRecord(1, watched("dead/a")), + fileRecord(2, watched("dead/b")), + fileRecord(3, watched("ready")), + fileRecord(4, watched("later")), + ]); + const exists = (path: string) => + fs.exists(path).pipe( + Effect.orDie, + Effect.map((found): Judgement => + found + ? { type: "met", outcome: "exists", detail: "It exists." } + : { type: "notYet" }, + ), + ); + // A new service on whatever store is there now (never seeded again): a restart. + const start = Effect.gen(function* () { + const childChecks: string[] = []; + const service = yield* makeService({ + filePath, + judges: { + file: (target) => + target.spec.path.includes("/dead/") + ? (Effect.never as Effect.Effect) + : exists(target.spec.path), + }, + fileChildJudge: (target) => + Effect.sync(() => childChecks.push(target.spec.path)).pipe( + Effect.andThen(exists(target.spec.path)), + ), + }); + return { service, childChecks }; + }); + const stateOf = (n: number) => + Effect.map( + storedWaits(filePath), + (waits) => waits.find((w) => w.id === idOf(n))?.state, + ); + + const before = yield* start; + // a and b hang; once they are late, ready and later are judged in their own processes. + assert.isTrue(yield* pass(before.service)); + yield* TestClock.adjust(5_000); + assert.isTrue(yield* pass(before.service)); + assert.deepStrictEqual(before.childChecks, [watched("ready"), watched("later")]); + assert.strictEqual(yield* stateOf(3), "delivering"); + assert.strictEqual(yield* stateOf(4), "waiting"); + const leftBehind = JSON.parse(yield* fs.readFileString(filePath)); + assertWallTimesOnly(leftBehind); + + // The server restarts on that store: its monotonic clock starts again, a day behind. + // Any monotonic time kept from the old run (a stuck check's lateAt) would now lie a day + // ahead, so a and b would never turn late and later would never be judged. + yield* fs.writeFileString(watched("later"), ""); + shift.monotonic = -8 * 24 * HOUR; + yield* TestClock.adjust(MINUTE); + const after = yield* start; + assert.isTrue(yield* pass(after.service)); + assert.strictEqual(yield* stateOf(4), "waiting"); + yield* TestClock.adjust(5_000); + assert.isTrue(yield* pass(after.service)); + assert.deepStrictEqual(after.childChecks, [watched("later")]); + assert.strictEqual(yield* stateOf(4), "delivering"); + assertWallTimesOnly(JSON.parse(yield* fs.readFileString(filePath))); + }).pipe(Effect.provideService(Clock.Clock, clock)); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); + + /** For the early-send tests: a met at once, b met after 3 s (both file), c a 55 s pr check. */ + const earlyWaits = [ + record({ + id: idOf(1), + key: "1".padStart(16, "0"), + spec: { path: watched("a"), until: "exists" }, + }), + record({ + id: idOf(2), + key: "2".padStart(16, "0"), + threadId: OTHER_THREAD, + spec: { path: watched("b"), until: "exists" }, + }), + record({ + id: idOf(3), + key: "3".padStart(16, "0"), + kind: "pr", + spec: { number: 1, until: "checks" }, + }), + ]; + const metJudgement: Judgement = { type: "met", outcome: "exists", detail: "It exists." }; + const earlyJudges: WaitJudges = { + file: (target) => + target.spec.path.endsWith("/b") + ? Effect.sleep(3_000).pipe(Effect.as(metJudgement)) + : Effect.succeed(metJudgement), + pr: () => Effect.sleep(55_000).pipe(Effect.as({ type: "notYet" })), + }; + + it.effect("two waits ending at different times in one pass: each is sent at once, and once", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + yield* seed(filePath, earlyWaits); + const tries: Array<[ThreadId, number]> = []; + const service = yield* makeService({ + filePath, + judges: earlyJudges, + delivery: { + receiver: () => Effect.succeed({ type: "ok" }), + deliver: (message) => + Effect.map(Clock.currentTimeMillis, (now): WaitDeliveryResult => { + tries.push([message.threadId, Math.floor((now - NOW) / 1_000)]); + return { type: "delivered" }; + }), + }, + }); + const sweep = yield* Effect.forkChild(service.runDue()); + yield* letFibersRun; + yield* stepSeconds(12); + assert.strictEqual(sweep.pollUnsafe(), undefined, "the pr check still runs"); + // Each within a second of its own answer, while the 55 s pr check still runs. + assert.deepStrictEqual(tries, [ + [THREAD, 0], + [OTHER_THREAD, 3], + ]); + yield* stepSeconds(70); + assert.isDefined(sweep.pollUnsafe()); + assert.strictEqual(tries.length, 2, "nothing sent twice"); + assert.deepStrictEqual( + (yield* storedWaits(filePath)).map((wait) => wait.state), + ["delivered", "delivered", "waiting"], + ); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); + + it.effect( + "stops half-way: a restart while an early send hangs sends the same message id, once", + () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + yield* seed(filePath, [earlyWaits[0], earlyWaits[2]]); + const ids: string[] = []; + const first = yield* makeService({ + filePath, + judges: earlyJudges, + delivery: { + receiver: () => Effect.succeed({ type: "ok" }), + deliver: (message) => + Effect.sync(() => void ids.push(message.messageId)).pipe( + Effect.andThen(Effect.never), + ), + }, + }); + const sweep = yield* Effect.forkChild(first.runDue()); + yield* stepSeconds(5); + assert.strictEqual(ids.length, 1, "the early send started"); + const mid = yield* storedWaits(filePath); + assert.strictEqual(mid[0]?.state, "delivering"); + assert.strictEqual(mid[0]?.messageId, ids[0], "the id is written before the send"); + // The server stops half-way through the pass. + yield* Fiber.interrupt(sweep); + const sent: string[] = []; + const second = yield* makeService({ + filePath, + judges: earlyJudges, + delivery: { + receiver: () => Effect.succeed({ type: "ok" }), + deliver: (message) => + Effect.sync((): WaitDeliveryResult => { + sent.push(message.messageId); + return { type: "delivered" }; + }), + }, + }); + const again = yield* Effect.forkChild(second.runDue()); + yield* stepSeconds(70); + assert.isDefined(again.pollUnsafe()); + assert.deepStrictEqual(sent, [ids[0]]); + assert.strictEqual((yield* storedWaits(filePath))[0]?.state, "delivered"); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); + + it.effect( + "a wake message that got 'try again later' is tried again after its delay even when the wall clock was set back an hour", + () => + Effect.gen(function* () { + const { clock, shift } = yield* shiftedClock; + yield* Effect.gen(function* () { + yield* TestClock.setTime(NOW); + yield* seed(filePath, [delivering({ deadlineAt: "2026-11-14T15:00:00.000Z" })]); + let busy = true; + let sent = 0; + const service = yield* makeService({ + filePath, + delivery: { + receiver: () => Effect.succeed({ type: "ok" }), + deliver: () => + Effect.sync((): WaitDeliveryResult => { + if (busy) return { type: "retry", detail: "busy" }; + sent += 1; + return { type: "delivered" }; + }), + }, + }); + assert.strictEqual((yield* service.deliverDue()).sent, 0); + busy = false; + // A time sync sets the wall clock back an hour; real time moves on two minutes, far + // past the retry delay (10 s). + shift.wall = -HOUR; + yield* TestClock.adjust(2 * MINUTE); + yield* service.deliverDue(); + assert.strictEqual( + sent, + 1, + "the retried message is held until the wall clock catches up", + ); + }).pipe(Effect.provideService(Clock.Clock, clock)); + }).pipe( + Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs), + Effect.provide(ServerSettingsService.layerTest(ON)), + ), + ); +}); diff --git a/apps/server/src/waits/WaitService.ts b/apps/server/src/waits/WaitService.ts index f4f5df35119c4..81181f0106222 100644 --- a/apps/server/src/waits/WaitService.ts +++ b/apps/server/src/waits/WaitService.ts @@ -59,6 +59,7 @@ import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as Latch from "effect/Latch"; import * as Option from "effect/Option"; import * as Semaphore from "effect/Semaphore"; @@ -122,6 +123,12 @@ const SWEEP_START_WINDOW_MS = 10_000; */ const FILE_CHILD_LATE_MS = FILE_CHILD_LIMIT_MS + 2_000; const SEND_CONCURRENCY = 4; +/** + * Now on the monotonic clock, in ms: when a check counts as late and how long a pass starts and + * waits, so a wall clock set back (a time sync, a restored snapshot) never stretches them. Only + * ever kept in memory, so a restart starts from nothing. + */ +const monotonicMillis = Effect.map(Clock.monotonicTimeNanos, (nanos) => Number(nanos / 1_000_000n)); /** * A send, or a look at a receiver, that has not answered by then counts as "try again later". The * late send is stopped in the background, and its message is not sent again while it still runs; @@ -330,8 +337,9 @@ interface RunningJudge extends JudgeRun { /** How long it may run before it counts as late. */ readonly limitMs: number; /** - * When it counts as late (stuck while it still runs), fixed when it starts: whoever started it - * may stop waiting for it at any time, even be cut off before it waits at all. + * When it counts as late (stuck while it still runs), on the monotonic clock + * (`monotonicMillis`), fixed when it starts: whoever started it may stop waiting for it at any + * time, even be cut off before it waits at all. */ readonly lateAt: number; } @@ -342,7 +350,7 @@ interface StartedJudge { } /** A started check; or none, because every place is held by a check that may still answer in time - * (`busy`, until `wakeAt` at the latest), or by late ones (`stuck`), or the thread already runs a + * (`busy`, until `wakeAt` at the latest, monotonic), or by late ones (`stuck`), or the thread already runs a * check (`threadBusy`). */ type JudgeStart = | { readonly type: "started"; readonly started: StartedJudge } @@ -444,7 +452,10 @@ export const make = Effect.fn("WaitService.make")(function* (input: { judgesChanged = yield* Deferred.make(); yield* Deferred.succeed(previous, undefined); }); - /** When a message is tried again, by message id; empty after a restart, so all go at once. */ + /** + * When a message is tried again, by message id, on the monotonic clock (a wall clock set back + * never holds a retry); empty after a restart, so all go at once. + */ const retryAt = new Map(); /** Tries of a message's current command id that recorded nothing, for its back-off. */ const failures = new Map(); @@ -452,6 +463,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const receiversToCheck = new Set(); /** The first sweep looks at every thread with a `waiting` wait (events missed while down). */ let checkAllReceivers = true; + /** When held notices are looked at again, on the monotonic clock; 0 when none are held. */ let noticesRetryAt = 0; /** Sends that have not returned, by message id, each with its own token. */ const sending = new Map(); @@ -684,18 +696,28 @@ export const make = Effect.fn("WaitService.make")(function* (input: { Effect.gen(function* () { const settings = yield* readSettingsOrNull; if (settings === null || !settings.waitFor.enabled) return; - const loaded = yield* ensureLoaded(true); - if (!loaded.ok) return; - const wait = loaded.contents.waits.find((candidate) => candidate.id === id); - if (wait === undefined || wait.state !== "waiting" || wait.kind !== pinned.kind) return; - if (waitKey(wait.threadId, targetOf(wait)) !== waitKey(wait.threadId, judged)) return; - const waits = loaded.contents.waits.map((candidate) => - candidate.id === id ? ({ ...candidate, spec: pinned.spec } as WaitRecord) : candidate, - ); - yield* commitWaits(loaded.contents, waits, "it is pinned at its next read.", settings); + yield* pinInStore(id, judged, pinned, settings); }), ); + /** `pinStored`'s write, under `lock`. */ + const pinInStore = Effect.fnUntraced(function* ( + id: string, + judged: WaitTarget, + pinned: WaitTarget, + settings: ServerSettings, + ) { + const loaded = yield* ensureLoaded(true); + if (!loaded.ok) return; + const wait = loaded.contents.waits.find((candidate) => candidate.id === id); + if (wait === undefined || wait.state !== "waiting" || wait.kind !== pinned.kind) return; + if (waitKey(wait.threadId, targetOf(wait)) !== waitKey(wait.threadId, judged)) return; + const waits = loaded.contents.waits.map((candidate) => + candidate.id === id ? ({ ...candidate, spec: pinned.spec } as WaitRecord) : candidate, + ); + yield* commitWaits(loaded.contents, waits, "it is pinned at its next read.", settings); + }); + /** The gate let a holder start its agent: its wait's `reservedUntil` is cleared (row 15). */ const useStoredSlot = (key: string) => lock.withPermits(1)( @@ -816,7 +838,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { return yield* Effect.gen(function* () { // Waits for a free place within its limit; with none, the next sweep judges it first. - const until = now + REGISTER_JUDGE_LIMIT_MS; + const until = (yield* monotonicMillis) + REGISTER_JUDGE_LIMIT_MS; const context: WaitJudgeContext = { waitId: null, now, @@ -874,6 +896,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const contents = yield* loadedOrFail(true); const again = yield* findOrRefuse(contents); if (Option.isSome(again)) return waitingResult(again.value, true); + const builtWith = pinned; const record = { id: waitId, threadId, @@ -896,6 +919,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (started !== null && Option.isNone(answer)) { // Still running in time: it is the wait's one run, and the next pass takes its answer. adopted = true; + // A pin the run made while the wait was being written is saved now, not lost until + // its next check, which might judge a changed remote. + if (pinned !== null && pinned !== builtWith) { + yield* pinInStore(waitId, target, pinned, settings); + } yield* carry(waitId, started); } else if (started !== null) { // Not checked here: the next sweep judges it, and before the waits it judged last time. @@ -1068,7 +1096,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { ): Effect.Effect => Effect.uninterruptible( Effect.gen(function* () { - const now = yield* Clock.currentTimeMillis; + const now = yield* monotonicMillis; const set = runningOf(pool); if (pool === "check" && [...set].some((entry) => entry.threadId === run.threadId)) { return { type: "threadBusy" } as const; @@ -1124,9 +1152,9 @@ export const make = Effect.fn("WaitService.make")(function* (input: { }), ); - /** The check's answer by `until` (epoch ms), or none; the check itself runs on either way. */ + /** The check's answer by `until` (monotonic), or none; the check itself runs on either way. */ const awaitJudge = (started: StartedJudge, until: number) => - Effect.flatMap(Clock.currentTimeMillis, (now) => + Effect.flatMap(monotonicMillis, (now) => Fiber.join(started.fiber).pipe( Effect.timeoutOption(Duration.millis(Math.max(0, until - now))), ), @@ -1139,7 +1167,9 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const carry = Effect.fnUntraced(function* (waitId: string, started: StartedJudge) { const kept: CarriedJudge = { entry: started.entry, answer: null }; carried.set(waitId, kept); - nextCheckAt.set(waitId, started.entry.lateAt); + // When it turns late, on the wall clock that times the waits. + const left = Math.max(0, started.entry.lateAt - (yield* monotonicMillis)); + nextCheckAt.set(waitId, (yield* Clock.currentTimeMillis) + left); yield* Effect.forkIn( Fiber.join(started.fiber).pipe( Effect.flatMap((judgement) => @@ -1200,7 +1230,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { }); /** - * At registration: a check started within `until` (epoch ms), waiting for a free slot while one + * At registration: a check started within `until` (monotonic), waiting for a free slot while one * may still free up in time; null when the checks are stuck, the thread runs a check already, or * the time is up. */ @@ -1217,7 +1247,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const result = yield* tryStartJudge(kind, run, judge, childJudge, limitMs); if (result.type === "started") return result.started; if (result.type === "stuck" || result.type === "threadBusy") return null; - const now = yield* Clock.currentTimeMillis; + const now = yield* monotonicMillis; if (now >= until) return null; // Woken when a check returns, or when the first running one turns late. yield* Deferred.await(changed).pipe( @@ -1309,7 +1339,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { carried.delete(wait.id); return answered(yield* confirmMet(wait, waiting.answer, settings)); } - if (now < waiting.entry.lateAt) return { type: "skip" } as const; + if ((yield* monotonicMillis) < waiting.entry.lateAt) return { type: "skip" } as const; carried.delete(wait.id); return answered( lateJudgement( @@ -1504,9 +1534,9 @@ export const make = Effect.fn("WaitService.make")(function* (input: { * Due when its next check time has come, or lies more than one interval ahead: the wall clock * was set back (a time sync or a restored snapshot), which must not pause judging. */ - const isDue = (wait: WaitRecord, now: number, settings: ServerSettings) => { + const isDue = (wait: WaitRecord, now: number, monoNow: number, settings: ServerSettings) => { const kept = carried.get(wait.id); - if (kept !== undefined) return kept.answer !== null || now >= kept.entry.lateAt; + if (kept !== undefined) return kept.answer !== null || monoNow >= kept.entry.lateAt; const next = nextCheckAt.get(wait.id); return ( next === undefined || @@ -1566,9 +1596,14 @@ export const make = Effect.fn("WaitService.make")(function* (input: { { discard: true }, ); - const judgeSweep = Effect.fnUntraced(function* (settings: ServerSettings) { + /** `onEnded` runs each time an answer that ends its wait was saved, while the pass goes on. */ + const judgeSweep = Effect.fnUntraced(function* ( + settings: ServerSettings, + onEnded: Effect.Effect = Effect.void, + ) { yield* stopRemovedChecks(settings); const startedAt = yield* Clock.currentTimeMillis; + const monoStartedAt = yield* monotonicMillis; const due = yield* lock.withPermits(1)( Effect.gen(function* () { const loaded = yield* ensureLoaded(true); @@ -1579,7 +1614,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (wait.state !== "waiting") continue; waiting.add(wait.id); // Slot waits are granted by their own step, oldest first. - if (wait.kind !== "slot" && isDue(wait, startedAt, settings)) due.push(wait); + if (wait.kind !== "slot" && isDue(wait, startedAt, monoStartedAt, settings)) + due.push(wait); } for (const id of nextCheckAt.keys()) if (!waiting.has(id)) nextCheckAt.delete(id); yield* stopEndedRuns; @@ -1595,8 +1631,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { // is full, and starts nothing after `SWEEP_START_WINDOW_MS` (those waits go first next pass). // An answer that ends its wait is saved at once; the others in one write at the end. A late // answer is ignored and its wait judged again later. - const startUntil = startedAt + SWEEP_START_WINDOW_MS; - const answerUntil = startedAt + SWEEP_ANSWER_WINDOW_MS; + const startUntil = monoStartedAt + SWEEP_START_WINDOW_MS; + const answerUntil = monoStartedAt + SWEEP_ANSWER_WINDOW_MS; const taken = new Set(); const later: Array = []; let judged = 0; @@ -1606,7 +1642,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const takeNext = Effect.gen(function* () { for (;;) { // Past the window nothing new starts, not even in a slot that has just been freed. - if ((yield* Clock.currentTimeMillis) >= startUntil) return null; + if ((yield* monotonicMillis) >= startUntil) return null; const changed = judgesChanged; let wakeAt = Number.POSITIVE_INFINITY; let left = false; @@ -1633,7 +1669,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { wakeAt = Math.min(wakeAt, start.wakeAt); } if (!left) return null; - const now = yield* Clock.currentTimeMillis; + const now = yield* monotonicMillis; // Woken when a check returns, or when the first running one turns late. yield* Deferred.await(changed).pipe( Effect.timeoutOption(Duration.millis(Math.min(startUntil, wakeAt) - now)), @@ -1655,8 +1691,10 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (judgement === null) continue; judged += 1; const saved = yield* saveIfEnds(wait.id, judgement, settings); - if (saved === "ended") ended += 1; - else if (saved === "later") later.push([wait.id, judgement]); + if (saved === "ended") { + ended += 1; + yield* onEnded; + } else if (saved === "later") later.push([wait.id, judgement]); } }); yield* Effect.all( @@ -1928,6 +1966,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { // 2. Hand over the waits of gone receivers, pick the due messages, write their ids. const startedAt = yield* Clock.currentTimeMillis; + const monoStartedAt = yield* monotonicMillis; const planned = yield* lock.withPermits(1)( Effect.gen(function* () { const loaded = yield* ensureLoaded(true); @@ -1946,7 +1985,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { ); }); const due = deliveryGroups(waits).filter( - (group) => (retryAt.get(group.messageId) ?? 0) <= startedAt, + (group) => (retryAt.get(group.messageId) ?? 0) <= monoStartedAt, ); const fresh = new Map(); for (const group of due) { @@ -2023,6 +2062,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const loaded = yield* ensureLoaded(true); if (!loaded.ok) return { sent: 0, ended: 0 }; const now = yield* Clock.currentTimeMillis; + const monoNow = yield* monotonicMillis; const answers = new Map(); for (const [group, result] of results) { for (const wait of group.waits) answers.set(wait.id, result); @@ -2083,7 +2123,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { for (const [group, result] of results) { const at = nextTry(group, result, waits, now, settings); if (at === null) forgetMessage(group.messageId); - else retryAt.set(group.messageId, at); + // Kept as the time left, on the monotonic clock. + else retryAt.set(group.messageId, monoNow + Math.max(0, at - now)); } const sent = results.filter(([, result]) => result.type === "delivered").length; const ended = transitions.filter((transition) => @@ -2227,7 +2268,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { break; } } - noticesRetryAt = held ? (yield* Clock.currentTimeMillis) + PROBLEM_RECHECK_MS : 0; + noticesRetryAt = held ? (yield* monotonicMillis) + PROBLEM_RECHECK_MS : 0; if (done.size > 0 || replaced.size > 0) { yield* lock.withPermits(1)( Effect.gen(function* () { @@ -2256,14 +2297,18 @@ export const make = Effect.fn("WaitService.make")(function* (input: { return Option.getOrElse(result, () => 0); }); - /** When the next sweep has work, from what is in memory; Infinity when nothing is open. */ - const nextDueAt = (settings: ServerSettings, now: number) => { + /** + * When the next sweep has work, on the wall clock, from what is in memory; Infinity when nothing + * is open. Monotonic times (`monoNow`) are turned into the time left from `now`. + */ + const nextDueAt = (settings: ServerSettings, now: number, monoNow: number) => { if (current === null) return now + PROBLEM_RECHECK_MS; let next = Infinity; const at = (ms: number) => { next = Math.min(next, ms); }; - if (current.notices.length > 0) at(Math.max(noticesRetryAt, now)); + const fromMono = (ms: number) => now + Math.max(0, ms - monoNow); + if (current.notices.length > 0) at(fromMono(noticesRetryAt)); if (receiversToCheck.size > 0) at(now); // A held slot that ends may go to a slot wait still waiting. const slotsWaiting = current.waits.some( @@ -2285,7 +2330,10 @@ export const make = Effect.fn("WaitService.make")(function* (input: { ); } } - for (const group of deliveryGroups(current.waits)) at(retryAt.get(group.messageId) ?? now); + for (const group of deliveryGroups(current.waits)) { + const retry = retryAt.get(group.messageId); + at(retry === undefined ? now : fromMono(retry)); + } return Math.max(next, now); }; @@ -2314,8 +2362,47 @@ export const make = Effect.fn("WaitService.make")(function* (input: { }); /** - * One sweep: deliver, judge, deliver what that ended, prune, notices; returns when the next one - * has work. Settings are read again before each step, so "off" stops it at the next step; a + * Runs `step` while sending the message of each wait it ends as soon as that is saved (a deliver + * step after each, one at a time), so a met wait's message never waits for slow checks. + * Returns once `step` and the last of those deliver steps are done. + */ + const withEarlyDelivery = Effect.fnUntraced(function* ( + step: (onEnded: Effect.Effect) => Effect.Effect, + ) { + let pending = false; + let finished = false; + const wake = yield* Latch.make(false); + const none = { sent: 0, ended: 0 }; + const deliverer = yield* Effect.forkChild( + Effect.gen(function* () { + for (;;) { + if (!pending) { + if (finished) return; + yield* wake.await; + yield* wake.close; + continue; + } + pending = false; + // Waits were switched off: the step after this one stops the sweep. + if ((yield* phase("deliver", deliverSweep, none)) === null) return; + } + }), + ); + const result = yield* step( + Effect.suspend(() => { + pending = true; + return Effect.asVoid(wake.open); + }), + ); + finished = true; + yield* wake.open; + yield* Fiber.join(deliverer); + return result; + }); + + /** + * One sweep: deliver, judge (sending each message it ends at once), deliver what that ended, + * prune, notices; returns when the next one has work. Settings are read again before each step, so "off" stops it at the next step; a * failing step is logged and skipped. It never overlaps itself. */ const runDue = Effect.fn("WaitService.runDue")(function* () { @@ -2334,7 +2421,9 @@ export const make = Effect.fn("WaitService.make")(function* (input: { Effect.gen(function* () { const none = { sent: 0, ended: 0 }; if ((yield* phase("deliver", deliverSweep, none)) === null) return stopped; - const judged = yield* phase("judge", judgeSweep, { judged: 0, ended: 0 }); + const judged = yield* withEarlyDelivery((onEnded) => + phase("judge", (latest) => judgeSweep(latest, onEnded), { judged: 0, ended: 0 }), + ); if (judged === null) return stopped; const granted = yield* phase("grant", grantSweep, 0); if (granted === null) return stopped; @@ -2345,8 +2434,9 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if ((yield* phase("notices", noticesSweep, 0)) === null) return stopped; const latest = (yield* readSettingsOrNull) ?? settings; const now = yield* Clock.currentTimeMillis; + const monoNow = yield* monotonicMillis; return yield* lock - .withPermits(1)(Effect.sync(() => ({ nextDueAt: nextDueAt(latest, now) }))) + .withPermits(1)(Effect.sync(() => ({ nextDueAt: nextDueAt(latest, now, monoNow) }))) .pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterruptsOnly(cause), diff --git a/apps/web/src/components/settings/WaitsAndSlotsSettings.browser.test.tsx b/apps/web/src/components/settings/WaitsAndSlotsSettings.browser.test.tsx new file mode 100644 index 0000000000000..2d6e694befb77 --- /dev/null +++ b/apps/web/src/components/settings/WaitsAndSlotsSettings.browser.test.tsx @@ -0,0 +1,166 @@ +import { DEFAULT_SERVER_SETTINGS, type ManagedPolicySummary } from "@t3tools/contracts"; +import { describe, expect, it, vi } from "vite-plus/test"; +import { page, userEvent } from "vite-plus/test/browser"; + +import "~/index.css"; +import { act, render } from "../../../test/render"; + +type Settings = typeof DEFAULT_SERVER_SETTINGS; +type UpdateOptions = { readonly onFailed?: () => void; readonly onSaved?: () => void }; + +const state = vi.hoisted(() => ({ + settings: {} as Settings, + environments: [] as Array<{ + environmentId: string; + providers: Array<{ instanceId: string; displayName: string }>; + managedPolicy?: ManagedPolicySummary; + }>, + update: vi.fn((_patch: unknown, _options?: UpdateOptions) => {}), +})); + +function scope() { + const environments = state.environments.map(({ environmentId, managedPolicy }) => ({ + environmentId, + label: environmentId, + connection: { phase: "connected" }, + serverConfig: { + settings: state.settings, + environment: { capabilities: {}, platform: { os: "linux", arch: "x64", machine: "server" } }, + ...(managedPolicy ? { managedPolicy } : {}), + }, + })); + return { + scope: { + kind: "all", + label: "All", + members: [], + environmentIds: environments.map(({ environmentId }) => environmentId), + }, + environments, + connectedEnvironments: environments, + environment: environments[0] ?? null, + targets: environments.map(({ environmentId }) => ({ + environmentId, + label: environmentId, + projectId: null, + settings: state.settings, + sources: {}, + })), + target: null, + }; +} + +vi.mock(import("./SettingsScopeContext"), async (original) => ({ + ...(await original()), + useSettingsScope: scope as never, + useOptionalSettingsScope: scope as never, +})); +vi.mock(import("./useScopedSettings"), async (original) => ({ + ...(await original()), + useScopedSettings: ((select: (settings: Settings) => unknown) => select(state.settings)) as never, + useScopedSettingsMixed: () => false, + useScopedSettingsWriteAllowed: () => true, + useUpdateScopedSettings: (() => state.update) as never, + useClearScopedSettings: (() => vi.fn()) as never, + useClearProjectOverrides: (() => vi.fn()) as never, +})); +vi.mock(import("../../state/environments"), async (original) => ({ + ...(await original()), + useEnvironmentProviders: (() => + new Map( + state.environments.map(({ environmentId, providers }) => [environmentId, providers]), + )) as never, +})); +vi.mock(import("../../state/session"), async (original) => ({ + ...(await original()), + useEnvironmentsWithScope: ((environments: ReadonlyArray<{ environmentId: string }>) => + new Set(environments.map(({ environmentId }) => environmentId))) as never, +})); + +import { WaitsAndSlotsSettings } from "./WaitsAndSlotsSettings"; + +const buildDone = { + argv: ["test", "-f", "/work/done"], + maxArgs: 0, + argPattern: "^[^-=@+]", + intervalSeconds: 30, + timeoutSeconds: 20, + description: "", +}; + +function setup(environments: (typeof state)["environments"]) { + state.settings = { + ...DEFAULT_SERVER_SETTINGS, + waitChecks: { "build-done": buildDone }, + } as Settings; + state.environments = environments; + state.update.mockReset(); +} + +const onlyEnvironment = [ + { environmentId: "env-a", providers: [{ instanceId: "claude-a", displayName: "Claude A" }] }, +]; + +describe("WaitsAndSlotsSettings (#105)", () => { + it("lists every selected environment's accounts, not only the primary's", async () => { + setup([ + ...onlyEnvironment, + { environmentId: "env-b", providers: [{ instanceId: "codex-b", displayName: "Codex B" }] }, + ]); + await using screen = await render(); + + await expect.element(screen.getByText("Agents on Claude A")).toBeVisible(); + await expect.element(screen.getByText("Agents on Codex B")).toBeVisible(); + }); + + it("an administrator lock on waitChecks makes Edit and Remove inert, like Add check", async () => { + setup([ + { + ...onlyEnvironment[0]!, + managedPolicy: { + organization: "Example Corp", + lockedSettings: ["waitChecks"], + }, + }, + ]); + await using screen = await render(); + + const button = (name: string) => screen.getByRole("button", { name }).element(); + await expect.element(screen.getByText("build-done")).toBeVisible(); + expect(button("Add check").closest("[inert]")).not.toBeNull(); + expect(button("Edit check build-done").closest("[inert]")).not.toBeNull(); + expect(button("Remove check build-done").closest("[inert]")).not.toBeNull(); + }); + + it("keeps the check dialog and its draft open until the write is saved", async () => { + setup(onlyEnvironment); + await using screen = await render(); + + await userEvent.click(screen.getByRole("button", { name: "Edit check build-done" })); + const dialog = page.getByRole("dialog"); + await expect.element(dialog).toBeVisible(); + const description = dialog.getByLabelText("Description"); + await userEvent.type(description, "Waits for the build"); + await userEvent.click(dialog.getByRole("button", { name: "Save check" })); + + expect(state.update).toHaveBeenCalledTimes(1); + expect(state.update.mock.calls[0]![0]).toEqual({ + waitChecks: { "build-done": { ...buildDone, description: "Waits for the build" } }, + }); + // The server has not answered yet, so the dialog stays and cannot send twice. + await expect.element(dialog).toBeVisible(); + await expect.element(dialog.getByRole("button", { name: "Saving…" })).toBeDisabled(); + + // The server refuses the write: the typed draft is still there to try again. + await act(() => state.update.mock.calls[0]![1]!.onFailed!()); + await expect.element(dialog.getByRole("alert")).toBeVisible(); + expect(dialog.getByRole("alert").element().textContent).toContain("draft is kept"); + await expect.element(description).toHaveValue("Waits for the build"); + await userEvent.click(dialog.getByRole("button", { name: "Save check" })); + expect(state.update).toHaveBeenCalledTimes(2); + + // Saved: only now does the dialog close. + await act(() => state.update.mock.calls[1]![1]!.onSaved!()); + await expect.element(dialog).not.toBeInTheDocument(); + }); +}); diff --git a/apps/web/src/components/settings/WaitsAndSlotsSettings.tsx b/apps/web/src/components/settings/WaitsAndSlotsSettings.tsx index 857f5347f2f4d..23586c75f3cbe 100644 --- a/apps/web/src/components/settings/WaitsAndSlotsSettings.tsx +++ b/apps/web/src/components/settings/WaitsAndSlotsSettings.tsx @@ -216,14 +216,20 @@ function WaitCheckDialog({ readonly onOpenChange: (open: boolean) => void; readonly previousName: string | null; readonly initialDraft: WaitCheckDraft; - readonly onSave: (name: string, check: WaitCheck) => string | null; + /** + * Returns an error to show, or null once the write is sent. The caller closes the dialog when + * the write is saved and calls `onFailed` when it is not, so the typed draft stays. + */ + readonly onSave: (name: string, check: WaitCheck, onFailed: () => void) => string | null; }) { const baseId = useId(); const [draft, setDraft] = useState(initialDraft); const [errors, setErrors] = useState>>({}); const [saveError, setSaveError] = useState(null); + const [saving, setSaving] = useState(false); const save = () => { + if (saving) return; const parsed = parseWaitCheckDraft(draft); if (!parsed.ok) { setErrors(parsed.errors); @@ -231,12 +237,16 @@ function WaitCheckDialog({ return; } setErrors({}); - const error = onSave(parsed.name, parsed.check); + setSaveError(null); + setSaving(true); + const error = onSave(parsed.name, parsed.check, () => { + setSaving(false); + setSaveError("The check was not saved. Your draft is kept, so you can try again."); + }); if (error !== null) { + setSaving(false); setSaveError(error); - return; } - onOpenChange(false); }; const field = (key: WaitCheckDraftField, multiline = false) => { @@ -315,7 +325,9 @@ function WaitCheckDialog({ - + @@ -339,13 +351,20 @@ function WaitChecksSection({ updateSettings }: { readonly updateSettings: Update setEditing({ previousName, draft, revision }); setRevision(revision + 1); }; - const save = (previousName: string | null) => (name: string, check: WaitCheck) => { - const result = withWaitCheck(checks, previousName, name, check); - if (!result.ok) return result.error; - // The server replaces the whole map, so the other checks are sent as they are. - updateSettings({ waitChecks: result.checks } satisfies ServerSettingsPatch); - return null; - }; + const save = + (previousName: string | null, dialogRevision: number) => + (name: string, check: WaitCheck, onFailed: () => void) => { + const result = withWaitCheck(checks, previousName, name, check); + if (!result.ok) return result.error; + // The server replaces the whole map, so the other checks are sent as they are. + updateSettings({ waitChecks: result.checks } satisfies ServerSettingsPatch, { + onFailed, + // Close only the dialog that sent this write, not one opened since. + onSaved: () => + setEditing((current) => (current?.revision === dialogRevision ? null : current)), + }); + return null; + }; return ( @@ -379,6 +398,9 @@ function WaitChecksSection({ updateSettings }: { readonly updateSettings: Update {name}} description={ <> @@ -428,7 +450,7 @@ function WaitChecksSection({ updateSettings }: { readonly updateSettings: Update }} previousName={editing.previousName} initialDraft={editing.draft} - onSave={save(editing.previousName)} + onSave={save(editing.previousName, editing.revision)} /> ) : null} @@ -437,11 +459,17 @@ function WaitChecksSection({ updateSettings }: { readonly updateSettings: Update function AgentSlotsSection({ updateSettings }: { readonly updateSettings: UpdateSettings }) { const slots = useScopedSettings((settings) => settings.agentSlots); - const { targets, environment } = useSettingsScope(); + const { targets, environment, connectedEnvironments } = useSettingsScope(); const providersByEnvironment = useEnvironmentProviders(); - const providers = - (environment ? providersByEnvironment.get(environment.environmentId) : undefined) ?? - EMPTY_SERVER_PROVIDERS; + // Every selected environment's accounts, the primary's first, so each can be limited here. + const providers = [ + ...(environment ? [environment] : []), + ...connectedEnvironments.filter( + (candidate) => candidate.environmentId !== environment?.environmentId, + ), + ].flatMap( + (candidate) => providersByEnvironment.get(candidate.environmentId) ?? EMPTY_SERVER_PROVIDERS, + ); const enabledMixed = targetsDiffer(targets, (settings) => settings.agentSlots.enabled); const accountsMixed = targetsDiffer(targets, (settings) => settings.agentSlots.maxPerAccount); const patch = (value: AgentSlotsSettingsPatch) => ({ agentSlots: value }); diff --git a/apps/web/src/components/settings/useScopedSettings.ts b/apps/web/src/components/settings/useScopedSettings.ts index 10ba059ceef81..4e7c90f6b9935 100644 --- a/apps/web/src/components/settings/useScopedSettings.ts +++ b/apps/web/src/components/settings/useScopedSettings.ts @@ -58,6 +58,7 @@ function useRunScopedPlan() { environments: Parameters[0], planner: Parameters[1], onFailed?: () => void, + onSaved?: () => void, ) => { const plan = planner(environments); if (plan.unavailableReason) { @@ -89,7 +90,10 @@ function useRunScopedPlan() { }, persistClientSettingsPatch, ).then(({ failedEnvironments, savedEnvironments, savedEnvironmentCount }) => { - if (failedEnvironments.length === 0) return; + if (failedEnvironments.length === 0) { + onSaved?.(); + return; + } onFailed?.(); toastManager.add({ type: "error", @@ -111,12 +115,19 @@ function useRunScopedPlan() { export function useUpdateScopedSettings() { const { scope, environments, deviceOnly } = useSettingsScope(); const run = useRunScopedPlan(); - /** `onFailed` runs when any target rejects or drops the patch, so a draft can fall back. */ - return (patch: ScopedSettingsPatch, options?: { readonly onFailed?: () => void }) => + /** + * `onFailed` runs when any target rejects or drops the patch, so a draft can fall back; + * `onSaved` runs once every target has saved it, so a form can close only then. + */ + return ( + patch: ScopedSettingsPatch, + options?: { readonly onFailed?: () => void; readonly onSaved?: () => void }, + ) => run( environments, (current) => planScopedSettingsPatch(scope, current, patch, { deviceOnly }), options?.onFailed, + options?.onSaved, ); } diff --git a/packages/shared/src/agentSlots.ts b/packages/shared/src/agentSlots.ts index 9ec5d0213b8fe..5b5bd3e3b8665 100644 --- a/packages/shared/src/agentSlots.ts +++ b/packages/shared/src/agentSlots.ts @@ -190,11 +190,37 @@ export interface SlotDecisionInput { readonly starting: ReadonlyArray; /** Live reservations that count (`liveReservations`), without the caller's own. */ readonly reservations?: ReadonlyArray; + /** + * Live held slots that count against their account only, never the machine: a thread's earlier + * hold while its re-ask is not kept yet (its new hold counts on the machine). Each thread counts + * once per account, and not where one of its `reservations` already holds that account. + */ + readonly accountOnly?: ReadonlyArray; readonly targetInstanceId: string; /** `null` when the memory floor is off. */ readonly freeMemoryBytes: number | null; } +/** Threads whose `accountOnly` holds are on `account` and not already counted there. */ +export function accountOnlyThreads( + counted: ReadonlyArray, + accountOnly: ReadonlyArray, + account: string, +): number { + const already = new Set( + counted.flatMap((reservation) => + reservation.account === account ? [reservation.threadId] : [], + ), + ); + return new Set( + accountOnly.flatMap((reservation) => + reservation.account === account && !already.has(reservation.threadId) + ? [reservation.threadId] + : [], + ), + ).size; +} + /** Machine limit first, then the target's account, then the memory floor. */ export function slotDecision(input: SlotDecisionInput): SlotDecision { const { limits, running, starting, targetInstanceId } = input; @@ -215,7 +241,8 @@ export function slotDecision(input: SlotDecisionInput): SlotDecision { const accountUsed = running.filter((agent) => agent.instanceId === targetInstanceId).length + starting.filter((entry) => entry.instanceId === targetInstanceId).length + - reservations.filter((reservation) => reservation.account === targetInstanceId).length; + reservations.filter((reservation) => reservation.account === targetInstanceId).length + + accountOnlyThreads(reservations, input.accountOnly ?? [], targetInstanceId); if (accountUsed >= accountMax) return refuse("account", accountUsed, accountMax); } if (