diff --git a/apps/server/src/slots/SlotGate.test.ts b/apps/server/src/slots/SlotGate.test.ts index d57957a27ff2..ea037a215420 100644 --- a/apps/server/src/slots/SlotGate.test.ts +++ b/apps/server/src/slots/SlotGate.test.ts @@ -900,6 +900,7 @@ describe("SlotGate", () => { assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), [ "granted", ]); + yield* gate.keep(["w1"]); assert.deepStrictEqual([...used].sort(), ["stored-0", "stored-1"]); assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 2 }); assert.strictEqual((yield* acquireAs(other)).type, "allow"); @@ -912,6 +913,92 @@ describe("SlotGate", () => { }), ); + // The old hold goes only once the new grant is written (`keep`). + it.effect( + "a re-ask that is not kept (cancelled, or its write failed) leaves the slot the thread held", + () => + Effect.gen(function* () { + const world = newWorld(); + const used: Array = []; + let inStore: ReadonlyArray = [ + { key: "stored", threadId: holder, untilMs: 50_000 }, + ]; + const attach = (gate: SlotGate.SlotGate["Service"]) => + gate.attachReservations({ + read: Effect.sync(() => inStore), + used: (key) => + Effect.sync(() => { + used.push(key); + inStore = inStore.filter((entry) => entry.key !== key); + }), + }); + yield* run( + world, + waitsOn({ maxPerMachine: 1, reserveSeconds: 60 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + yield* attach(gate); + assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), [ + "granted", + ]); + // Granted but not written yet: the stored hold is still there (a stop here keeps it). + assert.deepStrictEqual(used, []); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 1 }); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + // The re-ask was cancelled (or its write failed): its grant is dropped. + yield* gate.drop(["w1"]); + assert.deepStrictEqual(used, []); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 1 }); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + + // Asked again and written this time: the stored hold is replaced, still one slot. + assert.deepStrictEqual(types(yield* grant({ key: "w2", threadId: holder })), [ + "granted", + ]); + yield* gate.keep(["w2"]); + yield* gate.keep(["w2"]); + assert.deepStrictEqual(used, ["stored"]); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 1 }); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + // A late drop of a kept grant (or of one never made) changes nothing. + yield* gate.drop(["w1"]); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + + // Not kept, then its agent starts: the old and the new hold are both used up. + yield* TestClock.adjust("10 seconds"); + assert.deepStrictEqual(types(yield* grant({ key: "w3", threadId: holder })), [ + "granted", + ]); + assert.strictEqual((yield* acquireAs(holder)).type, "allow"); + yield* gate.drop(["w3"]); + assert.deepStrictEqual(used.slice(1).sort(), ["w2", "w3"]); + // Only the started agent counts now (its start grace), no held slot. + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 1 }); + }), + ); + // A memory-only hold: dropped re-ask, the hold stays until its own end. + yield* run( + world, + waitsOn({ maxPerMachine: 1, reserveSeconds: 60 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + assert.deepStrictEqual(types(yield* grant({ key: "m1", threadId: holder })), [ + "granted", + ]); + yield* gate.keep(["m1"]); + yield* TestClock.adjust("30 seconds"); + assert.deepStrictEqual(types(yield* grant({ key: "m2", threadId: holder })), [ + "granted", + ]); + yield* gate.drop(["m2"]); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + yield* TestClock.adjust("30 seconds"); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 0, max: 1 }); + }), + ); + }), + ); + 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 65fafadbc1de..6f8ad7a028e7 100644 --- a/apps/server/src/slots/SlotGate.ts +++ b/apps/server/src/slots/SlotGate.ts @@ -62,7 +62,7 @@ export type SlotGrant = | { readonly type: "off" } /** * Free: held for the thread until `untilMs` (memory only). A thread holds at most one slot: a - * slot it already held moves to this request (its end renewed), it is never added to. + * slot it already held moves to this request once the grant is kept (`keep`), never added to. */ | { readonly type: "granted"; readonly untilMs: number } /** An earlier request of the same thread was granted in this call: asked again next time. */ @@ -110,7 +110,15 @@ export class SlotGate extends Context.Service< readonly grant: ( requests: ReadonlyArray, ) => Effect.Effect>; - /** Grants that were not kept (the wait was cancelled, or its write failed) hold nothing. */ + /** + * Grants that were written: a slot their thread held before is used up now (memory, and + * cleared in the store). Until then a re-ask only hides it, so a stop half-way keeps it. + */ + readonly keep: (keys: ReadonlyArray) => Effect.Effect; + /** + * Grants that were not kept (the wait was cancelled, or its write failed) hold nothing; a slot + * the thread held before counts again. + */ readonly drop: (keys: ReadonlyArray) => Effect.Effect; /** For `wait_list`: null while slots are off, or when the running agents cannot be read. */ readonly usage: Effect.Effect; @@ -131,6 +139,13 @@ const make = Effect.gen(function* () { const held = yield* Ref.make>([]); /** 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. + */ + const replacing = new Map>(); + const hiddenKeys = () => + new Set([...replacing.values()].flatMap((olds) => olds.map((old) => old.key))); let stored: StoredReservations | null = null; /** @@ -181,11 +196,16 @@ const make = Effect.gen(function* () { */ const liveHeld = Effect.fn("SlotGate.liveHeld")(function* (waitsOn: boolean, nowMs: number) { for (const [key, untilMs] of usedKeys) if (untilMs <= nowMs) usedKeys.delete(key); + // A grant never kept nor dropped (a cut-off call) hides the old holds only until they end. + for (const [key, olds] of replacing) { + if (olds.every((old) => old.untilMs <= nowMs)) replacing.delete(key); + } const memory = yield* Ref.updateAndGet(held, (current) => liveReservations(current, nowMs)); if (!waitsOn) return []; const fromStore = stored === null ? [] : yield* stored.read; + const hidden = hiddenKeys(); return liveReservations([...memory, ...fromStore], nowMs).filter( - (reservation) => !usedKeys.has(reservation.key), + (reservation) => !usedKeys.has(reservation.key) && !hidden.has(reservation.key), ); }); @@ -197,21 +217,38 @@ const make = Effect.gen(function* () { /** Settings that cannot be read count as off, as for the usage pacing gate. */ const readSettings = serverSettings.getSettings.pipe(Effect.option); - /** - * Every slot held for `threadId` stops counting (memory ones go; stored ones are skipped until - * their end). Returns their keys, for the store to clear. Under the semaphore. - */ - const useUp = Effect.fnUntraced(function* ( + /** Used up: they stop counting (memory ones go; stored ones are skipped until their end). */ + const useUpAll = Effect.fn("SlotGate.useUpAll")(function* ( reservations: ReadonlyArray, - threadId: string, ) { - const own = reservations.filter((reservation) => reservation.threadId === threadId); - if (own.length === 0) return []; - for (const reservation of own) usedKeys.set(reservation.key, reservation.untilMs); - yield* Ref.update(held, (current) => current.filter((entry) => entry.threadId !== threadId)); - return own.map((reservation) => reservation.key); + if (reservations.length === 0) return []; + for (const reservation of reservations) usedKeys.set(reservation.key, reservation.untilMs); + const keys = new Set(reservations.map((reservation) => reservation.key)); + yield* Ref.update(held, (current) => current.filter((entry) => !keys.has(entry.key))); + return [...keys]; }); + /** The holds of `threadId` that a re-ask not kept yet hides; their entries go. */ + const takeHidden = (threadId: string) => { + const taken: Array = []; + for (const [key, olds] of replacing) { + if (!olds.some((old) => old.threadId === threadId)) continue; + taken.push(...olds); + replacing.delete(key); + } + return taken; + }; + + /** + * Every slot held for `threadId` (hidden ones too) is used up. Returns their keys, for the store + * to clear. Under the semaphore. + */ + const useUp = (reservations: ReadonlyArray, threadId: string) => + useUpAll([ + ...reservations.filter((reservation) => reservation.threadId === threadId), + ...takeHidden(threadId), + ]); + /** Clears used or replaced held slots in the store, outside the semaphore (see `acquire`). */ const clearStored = (keys: ReadonlyArray) => stored === null || keys.length === 0 @@ -293,16 +330,16 @@ const make = Effect.gen(function* () { return outcome; }); - const grantLocked = Effect.fn("SlotGate.grantLocked")( + const grant = Effect.fn("SlotGate.grant")( function* (requests: ReadonlyArray) { const settings = yield* readSettings; const limits = Option.isNone(settings) ? null : effectiveSlotLimits(settings.value.agentSlots); if (Option.isNone(settings) || limits === null) { - return { grants: requests.map((): SlotGrant => ({ type: "off" })), replaced: [] }; + return requests.map((): SlotGrant => ({ type: "off" })); } - if (requests.length === 0) return { grants: [], replaced: [] }; + if (requests.length === 0) return []; const nowMs = yield* Clock.currentTimeMillis; /** A wait without an account waits for the machine limit and the memory floor only. */ const limitsFor = (request: SlotRequest): SlotLimits => @@ -317,7 +354,6 @@ const make = Effect.gen(function* () { const reserveMs = settings.value.agentSlots.reserveSeconds * 1000; let reservations = yield* liveHeld(settings.value.waitFor.enabled, nowMs); const grants: Array = []; - const replaced: Array = []; const grantedThreads = new Set(); for (const request of requests) { // One wake per thread in a call: its other slot waits ask again next time. @@ -363,8 +399,13 @@ const make = Effect.gen(function* () { ...(request.account === undefined ? {} : { account: request.account }), untilMs, }; - // The slot it held before (memory or store) is replaced by this one. - replaced.push(...(yield* useUp(reservations, request.threadId))); + // The slot it held before (memory or store) is replaced by this one once it is kept; until + // then it is hidden, so a cancelled or unwritten re-ask gives it back. + const olds = [ + ...reservations.filter((entry) => entry.threadId === request.threadId), + ...takeHidden(request.threadId), + ]; + if (olds.length > 0) replacing.set(request.key, olds); // Counted for the rest of this call even when held 0 s: one free slot, one wake (R17). reservations = [ ...reservations.filter((entry) => entry.threadId !== request.threadId), @@ -375,18 +416,22 @@ const make = Effect.gen(function* () { grantedThreads.add(request.threadId); grants.push({ type: "granted", untilMs }); } - return { grants, replaced }; + return grants; }, // Under the same semaphore as `acquire`: a delegation never takes a slot being granted. semaphore.withPermits(1), ); - const grant = Effect.fn("SlotGate.grant")(function* (requests: ReadonlyArray) { - const { grants, replaced } = yield* grantLocked(requests); - // A replaced slot in the store is cleared like a used one, outside the semaphore. - yield* clearStored(replaced); - return grants; - }); + const keepLocked = Effect.fn("SlotGate.keepLocked")(function* (keys: ReadonlyArray) { + const replaced: Array = []; + for (const key of keys) { + const olds = replacing.get(key); + if (olds === undefined) continue; + replacing.delete(key); + replaced.push(...(yield* useUpAll(olds))); + } + return replaced; + }, semaphore.withPermits(1)); const usage = Effect.gen(function* () { const settings = yield* readSettings; @@ -440,8 +485,13 @@ const make = Effect.gen(function* () { stored = source; }), ), + // A replaced slot in the store is cleared like a used one, outside the semaphore. + keep: Effect.fn("SlotGate.keep")(function* (keys: ReadonlyArray) { + yield* clearStored(yield* keepLocked(keys)); + }), drop: Effect.fn("SlotGate.drop")(function* (keys: ReadonlyArray) { if (keys.length === 0) return; + for (const key of keys) replacing.delete(key); yield* Ref.update(held, (current) => current.filter((entry) => !keys.includes(entry.key))); }), started: Effect.fn("SlotGate.started")((token: SlotToken | null, childThreadId: ThreadId) => diff --git a/apps/server/src/waits/WaitJudges.ts b/apps/server/src/waits/WaitJudges.ts index 44fb0a9bcc6c..aacc8b641820 100644 --- a/apps/server/src/waits/WaitJudges.ts +++ b/apps/server/src/waits/WaitJudges.ts @@ -38,6 +38,11 @@ export interface WaitJudgeContext { * same project's check. Without it the check judge works it out itself. */ readonly place?: CheckPlace | undefined; + /** + * Fixes the wait's target to what this read resolved (a `pr` wait's repository and host), so + * later reads keep to it. Kept only while the wait is still waiting with the target judged. + */ + readonly pin?: ((target: WaitTarget) => Effect.Effect) | undefined; } export type WaitJudge = ( diff --git a/apps/server/src/waits/WaitPrJudge.test.ts b/apps/server/src/waits/WaitPrJudge.test.ts index cbf8dc93a977..65d2102f7be5 100644 --- a/apps/server/src/waits/WaitPrJudge.test.ts +++ b/apps/server/src/waits/WaitPrJudge.test.ts @@ -23,7 +23,9 @@ import * as Path from "effect/Path"; import { TestClock } from "effect/testing"; import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts"; +import * as ProjectFaviconResolver from "../project/ProjectFaviconResolver.ts"; import * as ProjectService from "../project/ProjectService.ts"; +import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { ServerSettingsService } from "../serverSettings.ts"; import type { WaitDelivery } from "./WaitDelivery.ts"; @@ -102,6 +104,7 @@ const fakeHost = () => { const prs = new Map(); const reads: string[] = []; const lookups: string[] = []; + const refreshes: string[] = []; const projects = new Map([ [PROJECT, shell(PROJECT, "acme/widgets")], [BARE_PROJECT, shell(BARE_PROJECT, null)], @@ -118,7 +121,7 @@ const fakeHost = () => { if (pr === undefined) return Effect.fail(providerError("not-found", "Could not resolve.")); return pr.error === undefined ? Effect.succeed(pr) : Effect.fail(pr.error); }); - const layer = Layer.mergeAll( + const services = Layer.mergeAll( Layer.mock(PullRequestService.PullRequestService)({ detail: (ref) => find(ref, "detail").pipe( @@ -144,12 +147,49 @@ const fakeHost = () => { Layer.mock(ProjectService.ProjectService)({ getShell: (projectId) => Effect.succeed(Option.fromNullishOr(projects.get(projectId))), }), + ); + const layer = Layer.merge( + services, Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({ readRepositoryIdentity: (workspaceRoot) => Effect.sync(() => (lookups.push(workspaceRoot), remotes.get(workspaceRoot) ?? null)), + refreshRepositoryIdentity: (workspaceRoot) => + Effect.sync(() => void refreshes.push(workspaceRoot)), }), ); - return { prs, reads, lookups, projects, remotes, layer }; + return { prs, reads, lookups, refreshes, projects, remotes, services, layer }; +}; + +/** + * T3's own project enrichment over a remote lookup that timed out once: like the real resolver, + * it then answers "none" from its cache for a minute unless asked to look again (`refresh`). + */ +const timedOutOnceRemote = (repository: string) => { + const calls: string[] = []; + let cachedNone = false; + const resolver = Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, { + resolve: (workspaceRoot, options) => + Effect.sync(() => { + const refresh = options?.refresh === true; + calls.push(refresh ? "refresh" : "read"); + if (calls.length === 1) { + cachedNone = true; + return null; + } + if (cachedNone && !refresh) return null; + cachedNone = false; + return { ...identityOf(repository), rootPath: workspaceRoot }; + }), + }); + const favicons = Layer.succeed(ProjectFaviconResolver.ProjectFaviconResolver, { + invalidate: () => Effect.void, + resolvePath: () => Effect.succeed(null), + }); + const layer = Layer.effect( + ProjectEnrichmentService.ProjectEnrichmentService, + ProjectEnrichmentService.make(), + ).pipe(Layer.provide(Layer.merge(resolver, favicons))); + return { calls, layer }; }; /** Records each message; every thread is there. */ @@ -421,7 +461,10 @@ it.layer(NodeServices.layer)("WaitPrJudge", (it) => { assert.notMatch(refused.message, /\bpass\b/iu); } assert.deepStrictEqual(host.reads, []); - assert.deepStrictEqual(host.lookups, [`/work/${BARE_PROJECT}`, `/work/${BARE_PROJECT}`]); + // Each new wait looked again past the cached "none" before it was refused. + const bare = `/work/${BARE_PROJECT}`; + assert.deepStrictEqual(host.lookups, [bare, bare, bare, bare]); + assert.deepStrictEqual(host.refreshes, [bare, bare]); yield* register(service, prWait({ number: 7, until: "merged" })); host.projects.delete(PROJECT); @@ -612,4 +655,187 @@ it.layer(NodeServices.layer)("WaitPrJudge", (it) => { ), ); }); + it.effect( + "a new wait looks past a timed-out remote lookup: it is not refused as having no remote", + () => { + const host = fakeHost(); + const remote = timedOutOnceRemote("acme/widgets"); + return Effect.gen(function* () { + yield* TestClock.setTime(NOW); + host.prs.set("acme/widgets#7", { state: "open", headSha: HEAD_A }); + host.projects.set(PROJECT, shell(PROJECT, null)); + // Something read the remote first and its `git` lookup timed out: "none" is cached. + const enrichment = yield* ProjectEnrichmentService.ProjectEnrichmentService; + assert.isNull(yield* enrichment.readRepositoryIdentity(`/work/${PROJECT}`)); + const filePath = yield* storePath; + const service = yield* makeService(filePath, yield* makePrJudge()); + + yield* TestClock.adjust(10_000); + const result = yield* register(service, prWait({ number: 7 })); + assert.strictEqual(result.state, "waiting"); + assert.deepStrictEqual(remote.calls, ["read", "refresh", "read"]); + assert.deepStrictEqual(host.reads.slice(0, 1), [ + "detail github.com acme/widgets#7 stale:false", + ]); + // Asked again: the same wait, and the remote, now known, is not looked up past the cache. + assert.strictEqual((yield* register(service, prWait({ number: 7 }))).state, "waiting"); + assert.deepStrictEqual(remote.calls, ["read", "refresh", "read"]); + }).pipe( + Effect.provide( + Layer.mergeAll(host.services, remote.layer, ServerSettingsService.layerTest(ON)), + ), + ); + }, + ); + + it.effect( + "a wait keeps the repository and host of its first read: a changed remote never ends it from another pull request", + () => { + const host = fakeHost(); + return Effect.gen(function* () { + yield* TestClock.setTime(NOW); + host.prs.set("acme/widgets#7", { state: "open", headSha: HEAD_A }); + const filePath = yield* storePath; + const service = yield* makeService(filePath, yield* makePrJudge()); + const made = yield* register(service, prWait({ number: 7, until: "merged" })); + assert.strictEqual(made.state, "waiting"); + const [stored] = yield* storedWaits(filePath); + assert.deepStrictEqual(stored?.spec, { + number: 7, + repository: "acme/widgets", + host: "github.com", + until: "merged", + }); + + // The project's remote flips to upstream, where the same number is merged. + 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 }); + yield* TestClock.adjust(SWEEP); + assert.deepStrictEqual(yield* service.judgeDue(), { judged: 1, ended: 0 }); + assert.isFalse(host.reads.some((read) => read.includes("upstream/"))); + // Asking the same again finds this wait (its key is the asked one), not a new one. + const again = yield* register(service, prWait({ number: 7, until: "merged" })); + assert.strictEqual( + again.state === "waiting" ? again.waitId : null, + made.state === "waiting" ? made.waitId : "", + ); + + host.prs.set("acme/widgets#7", { state: "merged", headSha: HEAD_A }); + yield* TestClock.adjust(SWEEP); + assert.deepStrictEqual(yield* service.judgeDue(), { judged: 1, ended: 1 }); + assert.strictEqual((yield* storedWaits(filePath))[0]?.result?.outcome, "merged"); + }).pipe(Effect.provide(Layer.merge(host.layer, ServerSettingsService.layerTest(ON)))); + }, + ); + + it.effect( + "a wait stored before the pin gets it at its first read after the upgrade, and keeps it after a restart", + () => { + const host = fakeHost(); + return Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const fs = yield* FileSystem.FileSystem; + host.prs.set("acme/widgets#7", { state: "open", headSha: HEAD_A }); + host.prs.set("acme/widgets#8", { state: "open", headSha: HEAD_A }); + const filePath = yield* storePath; + const before = yield* makeService(filePath, yield* makePrJudge()); + yield* register(before, prWait({ number: 7, until: "merged" })); + yield* register(before, prWait({ number: 8, repository: "Acme/Widgets", until: "merged" })); + // As an older build stored them: only what was asked, no host and no resolved repository. + const file = JSON.parse(yield* fs.readFileString(filePath)); + file.waits[0].spec = { number: 7, until: "merged" }; + file.waits[1].spec = { number: 8, repository: "Acme/Widgets", until: "merged" }; + yield* fs.writeFileString(filePath, JSON.stringify(file)); + + const after = yield* makeService(filePath, yield* makePrJudge()); + yield* TestClock.adjust(SWEEP); + assert.deepStrictEqual(yield* after.judgeDue(), { judged: 2, ended: 0 }); + assert.deepStrictEqual( + (yield* storedWaits(filePath)).map((wait) => wait.spec), + [ + { number: 7, repository: "acme/widgets", host: "github.com", until: "merged" }, + { number: 8, repository: "acme/widgets", host: "github.com", until: "merged" }, + ], + ); + + // Restart, and the remote flips: the pinned waits still read 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 }); + const restarted = yield* makeService(filePath, yield* makePrJudge()); + yield* TestClock.adjust(SWEEP); + assert.deepStrictEqual(yield* restarted.judgeDue(), { judged: 2, ended: 0 }); + assert.isFalse(host.reads.some((read) => read.includes("upstream/"))); + }).pipe(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* () { + yield* TestClock.setTime(NOW); + host.prs.set("acme/widgets#7", { state: "open", headSha: HEAD_A }); + const filePath = yield* storePath; + const fs = yield* FileSystem.FileSystem; + const first = yield* makeService(filePath, yield* makePrJudge()); + const made = yield* register(first, prWait({ number: 7, until: "merged" })); + const waitId = made.state === "waiting" ? made.waitId : ""; + // Stored by an older build, without the pin. + const file = JSON.parse(yield* fs.readFileString(filePath)); + file.waits[0].spec = { number: 7, until: "merged" }; + yield* fs.writeFileString(filePath, JSON.stringify(file)); + + // The read sees the wait waiting; it is cancelled before the read's pin is written. + const prJudge = yield* makePrJudge(); + const holder: { service?: WaitService.WaitService } = {}; + const cancelFirst: WaitJudge<"pr"> = (target, context) => + prJudge(target, { + ...context, + pin: (next) => + holder + .service!.cancel({ projectId: PROJECT, waitId }) + .pipe(Effect.ignore, Effect.andThen(context.pin!(next))), + }); + const service = yield* makeService(filePath, cancelFirst); + holder.service = service; + yield* TestClock.adjust(SWEEP); + yield* service.judgeDue(); + const [cancelled] = yield* storedWaits(filePath); + assert.strictEqual(cancelled?.state, "cancelled"); + assert.deepStrictEqual(cancelled?.spec, { number: 7, until: "merged" }); + }).pipe(Effect.provide(Layer.merge(host.layer, ServerSettingsService.layerTest(ON)))); + }); + // A remote name longer than the stored format allows (a project in deeply nested groups): + // the wait is not pinned, and it survives a restart as it did before pins existed. + it.effect( + "a wait whose remote repository is too long to store is not pinned and survives a restart", + () => { + const host = fakeHost(); + const long = `acme/${"group-".repeat(40)}widgets`; + return Effect.gen(function* () { + yield* TestClock.setTime(NOW); + host.projects.set(PROJECT, shell(PROJECT, long)); + host.prs.set(`${long.toLowerCase()}#7`, { state: "open", headSha: HEAD_A }); + const filePath = yield* storePath; + const before = yield* makeService(filePath, yield* makePrJudge()); + assert.strictEqual( + (yield* register(before, prWait({ number: 7, until: "merged" }))).state, + "waiting", + ); + assert.deepStrictEqual((yield* storedWaits(filePath))[0]?.spec, { + number: 7, + until: "merged", + }); + + const restarted = yield* makeService(filePath, yield* makePrJudge()); + yield* TestClock.adjust(SWEEP); + assert.deepStrictEqual(yield* restarted.judgeDue(), { judged: 1, ended: 0 }); + assert.strictEqual( + (yield* storedWaits(filePath)).filter((wait) => wait.state === "waiting").length, + 1, + ); + }).pipe(Effect.provide(Layer.merge(host.layer, ServerSettingsService.layerTest(ON)))); + }, + ); }); diff --git a/apps/server/src/waits/WaitPrJudge.ts b/apps/server/src/waits/WaitPrJudge.ts index 2dff14bf5a93..ecaac56a6b84 100644 --- a/apps/server/src/waits/WaitPrJudge.ts +++ b/apps/server/src/waits/WaitPrJudge.ts @@ -4,6 +4,7 @@ * settle rule's "seen" head is kept in memory only: after a restart it takes two fresh reads. */ import type { PullRequestRef } from "@t3tools/contracts"; +import { WaitPrSpec } from "@t3tools/contracts/waitFor"; import { type Judgement, judgePrChecks, type PrSeen } from "@t3tools/shared/waitFor"; import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider"; import * as Effect from "effect/Effect"; @@ -92,6 +93,8 @@ function failedRead( return transient(clip(`Pull request ${name} could not be read: ${error.detail}`)); } +const isPrSpec = Schema.is(WaitPrSpec); + export const makePrJudge = Effect.fn("WaitPrJudge.make")(function* () { const settingsService = yield* ServerSettingsService; const projects = yield* ProjectService.ProjectService; @@ -119,14 +122,29 @@ export const makePrJudge = Effect.fn("WaitPrJudge.make")(function* () { // `git remote`), as T3's pull request service does, so "not looked up yet" never reads as // "no remote". const shell = project.value.value; - const identity = + let identity = shell.repositoryIdentity ?? (yield* enrichment.readRepositoryIdentity(shell.workspaceRoot)); + // A cached "none" may be a `git remote` lookup that timed out, kept for a minute: a + // new wait looks again past that cache before it is refused. A stored wait only retries. + if (identity === null && registering) { + yield* enrichment.refreshRepositoryIdentity(shell.workspaceRoot); + identity = yield* enrichment.readRepositoryIdentity(shell.workspaceRoot); + } if (identity === null) return remoteProblem(NO_REMOTE, registering); const withRemote = { ...shell, repositoryIdentity: identity }; const repository = spec.repository ?? sourceControlDefinitions.repositorySelector(identity); if (repository === null) return remoteProblem(NO_REPOSITORY, registering); + // A wait keeps the host and repository of its first read: a later remote change (a + // fork's remote flipped to upstream) must not end it from another pull request's number. const target = yield* Effect.result( - resolveTarget({ repository, number: spec.number }, withRemote), + resolveTarget( + { + repository, + number: spec.number, + ...(spec.host === undefined ? {} : { host: spec.host }), + }, + withRemote, + ), ); if (Result.isFailure(target)) return targetProblem(target.failure, registering); const ref: PullRequestRef = { @@ -136,6 +154,16 @@ export const makePrJudge = Effect.fn("WaitPrJudge.make")(function* () { number: target.success.number, allowStale: false, }; + const pinnedSpec = { + ...spec, + repository: target.success.repository, + host: target.success.host, + }; + // Pinned only when it fits the stored format: a remote name too long to store stays + // unpinned (read as before) instead of making the stored wait unreadable after a restart. + if (spec.host === undefined && context.pin !== undefined && isPrSpec(pinnedSpec)) { + yield* context.pin({ kind: "pr", spec: pinnedSpec }); + } const name = `${ref.repository}#${ref.number}`; const result = yield* Effect.result(read(ref, spec.until === "checks")); if (Result.isFailure(result)) return failedRead(result.failure, name, registering); diff --git a/apps/server/src/waits/WaitService.ts b/apps/server/src/waits/WaitService.ts index 5a47055681b5..f4f5df35119c 100644 --- a/apps/server/src/waits/WaitService.ts +++ b/apps/server/src/waits/WaitService.ts @@ -653,6 +653,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { return { type: "notYet", detail: "Older slot waits go first." } satisfies Judgement; } const [grant] = yield* gate.grant([{ key: waitId, threadId, account: spec.account }]); + // Met at once, nothing to write: the slot the thread held before moves to this answer now. + if (grant?.type === "granted") yield* gate.keep([waitId]); return grant === undefined ? ({ type: "notYet" } satisfies Judgement) : slotJudgement(grant, settings, true); @@ -672,6 +674,28 @@ export const make = Effect.fn("WaitService.make")(function* (input: { }), ); + /** + * A judge fixed a stored wait's target (`WaitJudgeContext.pin`): written only while waits are on + * and the wait is still waiting with the target that was judged, so a cancel, an end or another + * pin in between wins. A failed write is pinned again at the next read. + */ + const pinStored = (id: string, judged: WaitTarget, pinned: WaitTarget) => + lock.withPermits(1)( + 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); + }), + ); + /** The gate let a holder start its agent: its wait's `reservedUntil` is cleared (row 15). */ const useStoredSlot = (key: string) => lock.withPermits(1)( @@ -787,11 +811,21 @@ export const make = Effect.fn("WaitService.make")(function* (input: { let started: StartedJudge | null = null; /** Set once the stored wait owns the run; until then the run ends with the call. */ let adopted = false; + /** The target the judge fixed (`WaitJudgeContext.pin`) before the wait was stored. */ + let pinned: WaitTarget | null = null; 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 context: WaitJudgeContext = { waitId: null, now, threadId, projectId }; + const context: WaitJudgeContext = { + waitId: null, + now, + threadId, + projectId, + // Stored with the wait; a run that answers after it was stored writes it then. + pin: (next) => + adopted ? pinStored(waitId, target, next) : Effect.sync(() => void (pinned = next)), + }; // A slot wait asks the gate instead, which holds a slot it grants at once. started = slotGate !== undefined @@ -851,7 +885,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { state: "waiting", attempt: 0, ...(parentThreadId === undefined ? {} : { parentThreadId }), - ...target, + // The key stays the asked target's, so asking the same again finds this wait. + ...(pinned ?? target), } as WaitRecord; yield* commitOrFail( { ...contents, waits: [...contents.waits, record] }, @@ -1292,6 +1327,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { now, threadId: wait.threadId, projectId: wait.projectId, + pin: (next) => pinStored(wait.id, target, next), }; let context = base; let checkRun: JudgeRun["check"]; @@ -1685,7 +1721,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { yield* gate.drop(freed.map(([id]) => id)); return 0; } - const granted = yield* applyGrants(new Map(freed), settings); + const kept = yield* applyGrants(new Map(freed), settings); + // Written: the slots their threads held before are used up now, outside the lock (the gate + // clears them in the store through it). + yield* gate.keep(kept); + const granted = kept.length; return ( granted + (yield* applyJudgements( @@ -1695,11 +1735,14 @@ export const make = Effect.fn("WaitService.make")(function* (input: { ); }); - /** Writes the slots granted; a grant not kept (cancelled meanwhile, or not written) holds nothing. */ + /** + * Writes the slots granted and returns the ids written; a grant not kept (cancelled meanwhile, + * or not written) holds nothing, and a slot its thread held before counts again. + */ const applyGrants = (granted: ReadonlyMap, settings: ServerSettings) => lock.withPermits(1)( Effect.gen(function* () { - if (granted.size === 0) return 0; + if (granted.size === 0) return []; const loaded = yield* ensureLoaded(true); const now = yield* Clock.currentTimeMillis; const changes = new Map(); @@ -1732,7 +1775,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { nextCheckAt.delete(id); yield* logTransition(transition); } - return kept.size; + return [...kept.keys()]; }), ); diff --git a/packages/contracts/src/waitFor.ts b/packages/contracts/src/waitFor.ts index 5062fdb00d48..d2597e857a7e 100644 --- a/packages/contracts/src/waitFor.ts +++ b/packages/contracts/src/waitFor.ts @@ -244,6 +244,7 @@ export const WaitUnitScope = Schema.Literals(["system", "user"]); const PrNumber = PositiveInt; const PrRepository = TrimmedNonEmptyString.check(Schema.isMaxLength(200)); +const PrHost = TrimmedNonEmptyString.check(Schema.isMaxLength(255)); const FilePath = Schema.String.check(Schema.isMinLength(1), Schema.isMaxLength(1000)); const UnitName = Schema.String.check(Schema.isMinLength(1), Schema.isMaxLength(200)); const CheckArg = Schema.String.check(Schema.isMaxLength(1000)); @@ -253,6 +254,12 @@ const SlotAccount = ProviderInstanceId; export const WaitPrSpec = Schema.Struct({ number: PrNumber, repository: Schema.optionalKey(PrRepository), + /** + * Stored only, never asked for: the pull request's host, written with its repository at the + * wait's first read, so a later change of the project's remote never moves the wait to another + * pull request. A wait without it (stored before it existed) gets both at its next read. + */ + host: Schema.optionalKey(PrHost), until: WaitPrUntil, }); export const WaitFileSpec = Schema.Struct({ path: FilePath, until: WaitFileUntil });