- commit
- 3367ad20993073b6c84b940acb24b360d02a41be
- parent
- 6b929a9d808c356035c07ebfee06102d8c396c47
- author
- Dami <dami.etoile@gmail.com>
| M | src/ | 2 | +- |
| M | src/ | 47 | +++--- |
| M | src/ | 234 | ++++++++++++++++-------------- |
| M | test/ | 10 | +- |
| M | test/ | 33 | ++--- |
| M | test/ | 30 | +++- |
@@ -17,7 +17,7 @@ import { marmotPrincipal, marmotShape } from "./marmot.ts";
1717 // writeGate is what every write must pass, whoever sends it: shape, bans, the1818 // relay's state, fuel, and the one-group rule. "" lets it through.1919 export function writeGate(relay: Relay, e: Event, conn: ConnState | null, t: number): string {20- if (relay.graspBusy) return "restricted: Git transaction in progress; retry the event";20+ if (relay.repositoryAccess.blocked) return "restricted: relay operation in progress; retry the event";2121 const graspError = graspGate(relay, e, conn?.host ?? "");2222 if (graspError) return graspError;2323 const siteError = checkSite(e);@@ -21,10 +21,12 @@ const answer = (reason: string, status: number) => new Response(reason, { status
2121 // isGitPath reserves malformed Git paths too, so every failure has CORS.2222 export const isGitPath = (url: URL) => /^\/(?:npub|prs\/)/.test(url.pathname) || /\.git(?:\/|$)/.test(url.pathname);2323 24+export const gitStoragePrefix = async (relay: Relay, repo: RepositoryAnnouncement) => `${relay.slug}/git/${await repositoryStoragePrefix(repositoryAddress(npubEncode(repo.owner), repo.identifier))}`;25+2426 // gitRepository records each R2 key's retained size, including orphan writes.2527 // Reservations precede PUT: an ambiguous response never makes bytes free.2628 export async function gitRepository(relay: Relay, repo: RepositoryAnnouncement): Promise<WalRepository> {27- const prefix = `${relay.slug}/git/${await repositoryStoragePrefix(repositoryAddress(npubEncode(repo.owner), repo.identifier))}`;29+ const prefix = await gitStoragePrefix(relay, repo);2830 const r2 = new R2ObjectStore(relay.media, { prefix, maxObjectBytes: 4 * 1024 * 1024 });2931 const store: ObjectStore = {3032 get: async (key) => r2.get(key),@@ -127,29 +129,28 @@ export async function grasp(relay: Relay, req: Request, url: URL): Promise<Respo
127129 const limited = relay.ipLimit(req.headers.get("x-relay-ip") || "unknown", req.method === "POST" ? "events" : "reqs");128130 if (limited) return answer(limited, 429);129131 if (relay.fuelStatus().outOfFuel) return answer("restricted: relay storage or fuel limit reached", 403);130- if (relay.graspBusy || relay.graspControls) return answer("restricted: Git transaction in progress; retry", 429);131- relay.graspBusy = true;132- try {133- if (parsed.endpoint === "root") {134- if (req.method !== "GET" && req.method !== "HEAD") return answer("invalid: method not allowed", 405);135- const coordinate = naddrEncode({ kind: KIND_REPO, pubkey: repo.owner, identifier: repo.identifier });136- const html = page(repo.identifier, `<h1>${esc(repo.identifier)}</h1><p>Git storage follows signed Nostr repository state.</p><p><a href="nostr:${coordinate}">Open in a Nostr Git client</a></p><pre>git clone ${esc(relay.webURL(url.host) + parsed.prefix)}</pre>`);137- return new Response(req.method === "HEAD" ? null : html, { headers: { ...CORS, "content-type": "text/html; charset=utf-8", "cache-control": "no-store" } });132+ return relay.repositoryAccess.run("git", async () => {133+ try {134+ if (parsed.endpoint === "root") {135+ if (req.method !== "GET" && req.method !== "HEAD") return answer("invalid: method not allowed", 405);136+ const coordinate = naddrEncode({ kind: KIND_REPO, pubkey: repo.owner, identifier: repo.identifier });137+ const html = page(repo.identifier, `<h1>${esc(repo.identifier)}</h1><p>Git storage follows signed Nostr repository state.</p><p><a href="nostr:${coordinate}">Open in a Nostr Git client</a></p><pre>git clone ${esc(relay.webURL(url.host) + parsed.prefix)}</pre>`);138+ return new Response(req.method === "HEAD" ? null : html, { headers: { ...CORS, "content-type": "text/html; charset=utf-8", "cache-control": "no-store" } });139+ }140+ const wal = await gitRepository(relay, repo);141+ const response = await wal.withReadSession(async (scopedWal) => {142+ const handler = createGitHandler(authorizedRepository(relay, repo, scopedWal), { prefix: parsed.prefix, authorizePush: () => true, observe: (event) => relay.meterBytes(event.requestBytes, event.responseBytes) });143+ return await handler(req);144+ });145+ const headers = new Headers(response.headers);146+ for (const [key, value] of Object.entries(CORS)) headers.set(key, value);147+ return new Response(response.body, { status: response.status, headers });148+ } catch (error) {149+ return answer(error instanceof LimitError ? "restricted: Git repository limit reached" : "error: Git repository unavailable", error instanceof LimitError ? 413 : 503);150+ } finally {151+ await relay.ensureAlarm(now() + 60);138152 }139- const wal = await gitRepository(relay, repo);140- const response = await wal.withReadSession(async (scopedWal) => {141- const handler = createGitHandler(authorizedRepository(relay, repo, scopedWal), { prefix: parsed.prefix, authorizePush: () => true, observe: (event) => relay.meterBytes(event.requestBytes, event.responseBytes) });142- return await handler(req);143- });144- const headers = new Headers(response.headers);145- for (const [key, value] of Object.entries(CORS)) headers.set(key, value);146- return new Response(response.body, { status: response.status, headers });147- } catch (error) {148- return answer(error instanceof LimitError ? "restricted: Git repository limit reached" : "error: Git repository unavailable", error instanceof LimitError ? 413 : 503);149- } finally {150- relay.graspBusy = false;151- await relay.ensureAlarm(now() + 60);152- }153+ }, () => answer("restricted: relay operation in progress; retry", 429));153154 }154155 155156 // graspTick expires purgatory and cleans one repository per alarm. Event@@ -35,6 +35,7 @@ import { groupFacts, handleGroupEvent, isGroupManagement, isNIP43Request } from
3535 import { markView, notePresence, viewsTick } from "./views.ts";3636 import { KIND_VANISH, KIND_AUTH, KIND_REPORT, KIND_NOSTR_CONNECT, KIND_GROUP_MEMBERS, KIND_GROUP_PINS, KIND_RELAY_DISCOVERY, KIND_MARMOT_GROUP } from "./kinds.ts";3737 import { marmotPrincipal } from "./marmot.ts";38+import { RepositoryAccess } from "./repository-access.ts";3839 3940 export interface Env {4041 RELAY: DurableObjectNamespace<Relay>;@@ -127,8 +128,7 @@ export class Relay extends DurableObject<Env> {
127128 succession = new Succession(this);128129 // A job round in flight, so overlapping alarms do not run two.129130 private working = false;130- graspBusy = false;131- graspControls = 0;131+ readonly repositoryAccess = new RepositoryAccess();132132 // Presence (views.ts): who wrote lately, in memory; the sockets are asked133133 // directly. A broadcast is at most one per PRESENCE_THROTTLE_S.134134 presenceActive = new Map<string, number>();@@ -449,7 +449,8 @@ export class Relay extends DurableObject<Env> {
449449 450450 // teardown deletes everything this relay holds and returns the name to451451 // unclaimed. Every socket is closed. Nothing is recoverable afterwards.452- async teardown() {452+ async teardown(): Promise<void> {453+ if (!this.repositoryAccess.owned) return this.repositoryAccess.run("teardown", () => this.teardown(), () => { throw new Error("relay operation in progress; retry teardown"); });453454 await forgetSites(this);454455 await forgetDomains(this);455456 for (const ws of this.ctx.getWebSockets()) {@@ -528,22 +529,24 @@ export class Relay extends DurableObject<Env> {
528529 // holder "" lets anyone claim; a pubkey reserves the claim. Returns "" or529530 // a reason.530531 async lease(name: string, host: string, until: number, holder: string): Promise<string> {531- if (!this.settings.isUnclaimed()) return "taken";532- if (this.slug !== name) {533- this.slug = name;534- await this.ctx.storage.put("slug", name);535- }536- const day = new Date(until * 1000).toISOString().slice(0, 10);537- this.settings.update({538- lease: { until, holder },539- writes: "open",540- reads: "open",541- name: "",542- description: `Temporary relay. Anyone can read and write here until ${day}; then everything on it is deleted. To keep it, claim it at https://${host}/ (sign once). Or claim a new name and pull from this one.`,543- });544- this.settings.setRetention(null, LEASE_RETENTION_DAYS);545- await this.ensureAlarm(until);546- return "";532+ return this.repositoryAccess.run("control", async () => {533+ if (!this.settings.isUnclaimed()) return "taken";534+ if (this.slug !== name) {535+ this.slug = name;536+ await this.ctx.storage.put("slug", name);537+ }538+ const day = new Date(until * 1000).toISOString().slice(0, 10);539+ this.settings.update({540+ lease: { until, holder },541+ writes: "open",542+ reads: "open",543+ name: "",544+ description: `Temporary relay. Anyone can read and write here until ${day}; then everything on it is deleted. To keep it, claim it at https://${host}/ (sign once). Or claim a new name and pull from this one.`,545+ });546+ this.settings.setRetention(null, LEASE_RETENTION_DAYS);547+ await this.ensureAlarm(until);548+ return "";549+ }, () => "restricted: relay operation in progress; retry");547550 }548551 549552 // ---- forking: a new name, this relay's events pulled into it ----@@ -580,14 +583,16 @@ export class Relay extends DurableObject<Env> {
580583 // adoptFrom is the forked side, reached over RPC only: a fresh lease581584 // takes the people and starts pulling from the source.582585 async adoptFrom(sourceURL: string, filter: PullFilter, people: { pubkey: string; name: string | null; note: string }[], sourceHost: string): Promise<string> {583- if (!this.settings.isLeased()) return "restricted: only a fresh lease can adopt";584- if (this.store.stats().events > 0) return "restricted: this lease already holds events";585- const day = new Date((this.settings.policy.lease?.until ?? 0) * 1000).toISOString().slice(0, 10);586- this.settings.update({ description: `Forked from ${sourceHost}. Temporary until ${day} unless claimed; then everything on it is deleted. Claim it to keep it.` });587- const t = now();588- for (const m of people) this.settings.upsertMember(m.pubkey, { name: m.name, note: m.note, via: "forked" }, t, true);589- const r = await this.addChecked({ kind: "pull", label: "pull", relays: [sourceURL], filter, every: 0, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, last: null });590- return typeof r === "string" ? r : "";586+ return this.repositoryAccess.run("control", async () => {587+ if (!this.settings.isLeased()) return "restricted: only a fresh lease can adopt";588+ if (this.store.stats().events > 0) return "restricted: this lease already holds events";589+ const day = new Date((this.settings.policy.lease?.until ?? 0) * 1000).toISOString().slice(0, 10);590+ this.settings.update({ description: `Forked from ${sourceHost}. Temporary until ${day} unless claimed; then everything on it is deleted. Claim it to keep it.` });591+ const t = now();592+ for (const m of people) this.settings.upsertMember(m.pubkey, { name: m.name, note: m.note, via: "forked" }, t, true);593+ const r = await this.addChecked({ kind: "pull", label: "pull", relays: [sourceURL], filter, every: 0, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, last: null });594+ return typeof r === "string" ? r : "";595+ }, () => "restricted: relay operation in progress; retry");591596 }592597 593598 // ---- jobs: pulls, backfills, rebroadcasts, once or standing ----@@ -760,16 +765,23 @@ export class Relay extends DurableObject<Env> {
760765 }761766 762767 async fetch(req: Request): Promise<Response> {763- this.touch();764- const name = req.headers.get("x-relay-name");765- if (name && name !== this.slug) {766- this.slug = name;767- await this.ctx.storage.put("slug", name);768- }769768 const url = new URL(req.url);770769 const completed = (response: Response) => featureOn(this.settings.policy, "grasp") ? graspCORS(response, req.method === "OPTIONS") : response;771- // Relays claimed before identities existed get theirs on first contact.772- if (this.settings.policy.owner && !this.identity.pubkey) await this.publishMembership();770+ const name = req.headers.get("x-relay-name");771+ const initializes = Boolean(name && name !== this.slug) || Boolean(this.settings.policy.owner && !this.identity.pubkey);772+ this.touch();773+ if (initializes) {774+ const initialized = await this.repositoryAccess.run("control", async () => {775+ if (name && name !== this.slug) {776+ this.slug = name;777+ await this.ctx.storage.put("slug", name);778+ }779+ // Relays claimed before identities existed get theirs on first contact.780+ if (this.settings.policy.owner && !this.identity.pubkey) await this.publishMembership();781+ return true;782+ }, () => false);783+ if (!initialized) return completed(new Response(JSON.stringify({ error: "restricted: relay operation in progress; retry" }), { status: 429, headers: { "content-type": "application/json", "access-control-allow-origin": "*", "retry-after": "1" } }));784+ }773785 // The local custom-domain record wins over a cached edge mapping, so774786 // changing a relay domain into a site never exposes the old relay door.775787 const domainSite = this.settings.policy.customHosts?.find((h) => h.host === url.hostname)?.site;@@ -784,9 +796,7 @@ export class Relay extends DurableObject<Env> {
784796 if (upgrade) return this.acceptWebSocket(req);785797 if (isGitPath(url)) return route(this, req, url);786798 if (req.method !== "GET" && req.method !== "HEAD" && req.method !== "OPTIONS") {787- if (this.graspBusy) return completed(new Response("restricted: Git transaction in progress; retry", { status: 429, headers: { "access-control-allow-origin": "*", "retry-after": "1" } }));788- this.graspControls++;789- try { return completed(await route(this, req, url)); } finally { this.graspControls--; }799+ return completed(await this.repositoryAccess.run("control", () => route(this, req, url), () => new Response(JSON.stringify({ error: "restricted: relay operation in progress; retry" }), { status: 429, headers: { "content-type": "application/json", "access-control-allow-origin": "*", "retry-after": "1" } })));790800 }791801 return completed(await route(this, req, url));792802 }@@ -970,14 +980,15 @@ export class Relay extends DurableObject<Env> {
970980 // acceptAny routes the special kinds (zap receipts, NIP-56 reports) and971981 // falls back to the ordinary write path. Callers broadcast when stored.972982 async acceptAny(e: Event, conn: ConnState): Promise<{ ok: boolean; msg: string; stored: boolean }> {973- if (this.graspBusy) return { ok: false, msg: "restricted: Git transaction in progress; retry the event", stored: false };974- if (e.kind === 9735) {975- const r = await acceptReceipt(this, e, conn.host);976- if (r) return r;977- }978- if (e.kind === KIND_REPORT) return this.acceptReport(e, conn.host);979- if (isGroupManagement(e.kind) && (hasTag(e, "h") || isNIP43Request(e.kind))) return this.acceptGroup(e, conn);980- return this.accept(e, conn);983+ return this.repositoryAccess.run("event", async () => {984+ if (e.kind === 9735) {985+ const r = await acceptReceipt(this, e, conn.host);986+ if (r) return r;987+ }988+ if (e.kind === KIND_REPORT) return this.acceptReport(e, conn.host);989+ if (isGroupManagement(e.kind) && (hasTag(e, "h") || isNIP43Request(e.kind))) return this.acceptGroup(e, conn);990+ return this.accept(e, conn);991+ }, () => ({ ok: false, msg: "restricted: relay operation in progress; retry the event", stored: false }));981992 }982993 983994 // acceptGroup handles NIP-29 joins, leaves and moderation: the common gate,@@ -1023,6 +1034,7 @@ export class Relay extends DurableObject<Env> {
10231034 // accept runs the write-side rules and stores the event. conn is null for10241035 // host-side publishes (the dashboard), which skip client policy.10251036 accept(e: Event, conn: ConnState | null): { ok: boolean; msg: string; stored: boolean } {1037+ if (!this.repositoryAccess.owned) return this.repositoryAccess.sync("event", () => this.accept(e, conn), () => ({ ok: false, msg: "restricted: relay operation in progress; retry the event", stored: false }));10261038 const p = this.settings.policy;10271039 const t = now();10281040 const no = (msg: string) => ({ ok: false, msg, stored: false });@@ -1087,74 +1099,72 @@ export class Relay extends DurableObject<Env> {
10871099 }10881100 10891101 async alarm() {1090- if (this.graspBusy) { await this.ctx.storage.setAlarm(Date.now() + 1000); return; }1091- this.graspControls++;1092- try {1093- this.touch();1094- const t = now();1095- // A lease that has run out is wiped whole: the name is free again.1096- if (this.settings.leaseExpired(t)) {1097- await this.teardown();1098- return;1099- }1100- const graspAt = await graspTick(this);1101- await this.syncSites();1102- await queueMirrors(this);1103- if (await this.jobsTick()) {1104- await this.ctx.storage.setAlarm(Date.now() + 250);1105- return;1106- }1107- this.flushUsage();1108- this.fuel.chargeStorage(t, this.eventBytes(), this.mediaBytes());1109- this.store.drain();1110- this.sweepRetention(t);1111- await viewsTick(this, t);1112- await this.publishDiscovery();1113- // Fuel notice: once when it turns low, then once a day while it stays low.1114- if (this.settings.policy.notify.fuel) {1115- const low = fuelLow(this.fuelStatus());1116- const last = (await this.ctx.storage.get<number>("fuel-low-at")) ?? 0;1117- if (low && t - last >= 86400) {1118- await this.ctx.storage.put("fuel-low-at", t);1119- await notify(this, "fuel", fuelText(this, this.fuelStatus()), "fuel on " + this.slug);1120- } else if (!low && last) await this.ctx.storage.delete("fuel-low-at");1121- }1122- // Succession: the dead-man's switch, checked once a day.1123- await this.succession.tick(t);1124- // ---- digest: one message a week on how the relay is doing ----1125- if (this.settings.policy.notify.digest && this.settings.policy.owner !== "") {1126- const last = (await this.ctx.storage.get<number>("lastDigest")) ?? 0;1127- if (last === 0) await this.ctx.storage.put("lastDigest", t);1128- else if (t - last >= DIGEST_DAYS * 86400 - 3600) {1129- await this.ctx.storage.put("lastDigest", t);1130- await notify(this, "digest", await digestText(this, last, t), "the week on " + this.slug);1102+ return this.repositoryAccess.run("alarm", async () => {1103+ this.touch();1104+ const t = now();1105+ // A lease that has run out is wiped whole: the name is free again.1106+ if (this.settings.leaseExpired(t)) {1107+ await this.teardown();1108+ return;11311109 }1132- }1133- // ---- end digest ----1134- const next = this.store.sweepExpired(t);1135- if (await this.syncSites()) {1136- await this.ctx.storage.setAlarm(Date.now() + 250);1137- return;1138- }1139- // ---- dumps: a scheduled JSONL of everything, into R2 (dumps.ts) ----1140- if (dumpDue(this, t)) {1141- try {1142- await writeDump(this, t);1143- } catch (err) {1144- console.log("dump failed: " + (err instanceof Error ? err.message : String(err)));1110+ const graspAt = await graspTick(this);1111+ await this.syncSites();1112+ await queueMirrors(this);1113+ if (await this.jobsTick()) {1114+ await this.ctx.storage.setAlarm(Date.now() + 250);1115+ return;11451116 }1146- }1147- // ---- end dumps ----1148- let at = t + 86400;1149- if (this.sql.exec(`SELECT 1 FROM site_mirror_queue LIMIT 1`).toArray().length) at = Math.min(at, t + 60);1150- if (next > 0 && next < at) at = next;1151- const lease = this.settings.policy.lease;1152- if (this.settings.isLeased() && lease && lease.until < at) at = lease.until;1153- const run = await this.nextJobRun();1154- if (run > 0 && run < at) at = Math.max(run, t + 1);1155- if (graspAt && graspAt < at) at = graspAt;1156- await this.ctx.storage.setAlarm(at * 1000 + 500);1157- } finally { this.graspControls--; }1117+ this.flushUsage();1118+ this.fuel.chargeStorage(t, this.eventBytes(), this.mediaBytes());1119+ this.store.drain();1120+ this.sweepRetention(t);1121+ await viewsTick(this, t);1122+ await this.publishDiscovery();1123+ // Fuel notice: once when it turns low, then once a day while it stays low.1124+ if (this.settings.policy.notify.fuel) {1125+ const low = fuelLow(this.fuelStatus());1126+ const last = (await this.ctx.storage.get<number>("fuel-low-at")) ?? 0;1127+ if (low && t - last >= 86400) {1128+ await this.ctx.storage.put("fuel-low-at", t);1129+ await notify(this, "fuel", fuelText(this, this.fuelStatus()), "fuel on " + this.slug);1130+ } else if (!low && last) await this.ctx.storage.delete("fuel-low-at");1131+ }1132+ // Succession: the dead-man's switch, checked once a day.1133+ await this.succession.tick(t);1134+ // ---- digest: one message a week on how the relay is doing ----1135+ if (this.settings.policy.notify.digest && this.settings.policy.owner !== "") {1136+ const last = (await this.ctx.storage.get<number>("lastDigest")) ?? 0;1137+ if (last === 0) await this.ctx.storage.put("lastDigest", t);1138+ else if (t - last >= DIGEST_DAYS * 86400 - 3600) {1139+ await this.ctx.storage.put("lastDigest", t);1140+ await notify(this, "digest", await digestText(this, last, t), "the week on " + this.slug);1141+ }1142+ }1143+ // ---- end digest ----1144+ const next = this.store.sweepExpired(t);1145+ if (await this.syncSites()) {1146+ await this.ctx.storage.setAlarm(Date.now() + 250);1147+ return;1148+ }1149+ // ---- dumps: a scheduled JSONL of everything, into R2 (dumps.ts) ----1150+ if (dumpDue(this, t)) {1151+ try {1152+ await writeDump(this, t);1153+ } catch (err) {1154+ console.log("dump failed: " + (err instanceof Error ? err.message : String(err)));1155+ }1156+ }1157+ // ---- end dumps ----1158+ let at = t + 86400;1159+ if (this.sql.exec(`SELECT 1 FROM site_mirror_queue LIMIT 1`).toArray().length) at = Math.min(at, t + 60);1160+ if (next > 0 && next < at) at = next;1161+ const lease = this.settings.policy.lease;1162+ if (this.settings.isLeased() && lease && lease.until < at) at = lease.until;1163+ const run = await this.nextJobRun();1164+ if (run > 0 && run < at) at = Math.max(run, t + 1);1165+ if (graspAt && graspAt < at) at = graspAt;1166+ await this.ctx.storage.setAlarm(at * 1000 + 500);1167+ }, async () => { await this.ctx.storage.setAlarm(Date.now() + 1000); });11581168 }11591169 11601170 // sweepRetention applies the owner's keep-for rules. Returns how many@@ -21,8 +21,14 @@ export async function rpc(host: string, sk: Uint8Array | null, method: string, .
2121 const payload = { method, params };2222 const headers: Record<string, string> = { "content-type": "application/nostr+json+rpc" };2323 if (sk) headers.authorization = await nip98(sk, url, "POST", payload);24- const resp = await SELF.fetch(url, { method: "POST", headers, body: JSON.stringify(payload) });25- return { status: resp.status, ...(await resp.json<any>()) };24+ // Background alarms may own admission between management calls. Retry only25+ // this transient refusal; quota, rate and inventory cooldown errors surface.26+ for (let attempt = 0; ; attempt++) {27+ const resp = await SELF.fetch(url, { method: "POST", headers, body: JSON.stringify(payload) });28+ const result = { status: resp.status, ...(await resp.json<any>()) };29+ if (result.status !== 429 || result.error !== "restricted: relay operation in progress; retry" || attempt >= 20) return result;30+ await sleep(50);31+ }2632 }2733 2834 // info reads the NIP-11 document.@@ -58,8 +58,8 @@ describe("GRASP coordination", () => {
5858 const active = mode === "http" ? relay.fetch(request()) : relay.alarm();5959 try {6060 await atRead;61- expect(relay.graspBusy).toBe(mode === "http");62- expect(relay.graspControls).toBe(mode === "alarm" ? 1 : 0);61+ expect(relay.repositoryAccess.busy).toBe(true);62+ expect(relay.repositoryAccess.kind).toBe(mode === "http" ? "git" : "alarm");6363 const before = gets;6464 const otherBefore = otherOperations;6565 const refused = await relay.fetch(request());@@ -69,27 +69,18 @@ describe("GRASP coordination", () => {
6969 const admission = await relay.acceptAny(state, relay.virtualConn(host, pk(owner)));7070 expect(gets).toBe(before);7171 expect(otherOperations).toBe(otherBefore);72- if (mode === "http") {73- expect(admission.ok).toBe(false);74- expect(admission.msg).toContain("Git transaction in progress");75- const management = await relay.fetch(new Request(`http://${host}/`, { method: "POST", body: "{}" }));76- expect(management.status).toBe(429);77- await management.arrayBuffer();78- await relay.alarm();79- expect(gets).toBe(before);80- expect(relay.graspBusy).toBe(true);81- expect(relay.graspControls).toBe(0);82- } else {83- // Controls blocks Git admission, but event writes can still alter84- // authority while alarm work awaits R2. It is not a general lock.85- expect(admission.ok).toBe(true);86- expect(relay.sql.exec("SELECT id FROM events WHERE id=?", state.id).toArray()).toHaveLength(1);87- expect(relay.graspControls).toBe(1);88- }72+ expect(admission.ok).toBe(false);73+ expect(admission.msg).toContain("relay operation in progress");74+ const management = await relay.fetch(new Request(`http://${host}/`, { method: "POST", body: "{}" }));75+ expect(management.status).toBe(429);76+ await management.arrayBuffer();77+ if (mode === "http") await relay.alarm();78+ expect(gets).toBe(before);79+ expect(relay.repositoryAccess.busy).toBe(true);80+ expect(relay.repositoryAccess.kind).toBe(mode === "http" ? "git" : "alarm");8981 release();9082 const response = await active;91- expect(relay.graspBusy).toBe(false);92- expect(relay.graspControls).toBe(0);83+ expect(relay.repositoryAccess.busy).toBe(false);9384 const completedGets = gets;9485 if (response) {9586 expect(response.status).toBe(outcome === "failure" ? 503 : 200);@@ -276,11 +276,10 @@ describe("GRASP", () => {
276276 277277 const stub = env.RELAY.getByName("grasp-pr-checkpoint");278278 await runInDurableObject(stub, async (relay) => {279- relay.graspBusy = true;280- try {279+ await relay.repositoryAccess.run("git", async () => {281280 const wal = await gitRepository(relay, storedRepository(relay, pk(owner), "correction")!);282281 expect((await wal.checkpoint()).changed).toBe(true);283- } finally { relay.graspBusy = false; }282+ }, () => { throw new Error("Git scope unexpectedly refused"); });284283 });285284 const metadata = await runInDurableObject(stub, async (relay) => {286285 const bucket = relay.media;@@ -305,15 +304,14 @@ describe("GRASP", () => {
305304 const requestId = "hidden-pr-correction";306305 expect(await (await receiveSettled(path, host, null, next.commitID, ref, next.pack, requestId)).text()).toContain(`ok ${ref}`);307306 const sequence = await runInDurableObject(stub, async (relay) => {308- relay.graspBusy = true;309- try {307+ return relay.repositoryAccess.run("git", async () => {310308 const wal = await gitRepository(relay, storedRepository(relay, pk(owner), "correction")!);311309 for (let i = 0; i < 130; i++) await wal.commit({ id: `later-${i}`, updates: [{ name: `refs/tags/later-${i}`, old: null, new: first.commitID }] });312310 const snapshot = await wal.load();313311 expect(snapshot.records).toHaveLength(1);314312 expect(snapshot.records[0].id).not.toBe(requestId);315313 return snapshot.sequence;316- } finally { relay.graspBusy = false; }314+ }, () => { throw new Error("Git scope unexpectedly refused"); });317315 });318316 // Pending work for another repository does not reload this one's packs.319317 const other = await WS.connect(host);@@ -493,16 +491,30 @@ describe("GRASP", () => {
493491 await rpc(host, owner, "setpolicy", { features: { grasp: true } });494492 const event = repository(owner, host, "fenced");495493 await runInDurableObject(env.RELAY.getByName("grasp-fence"), async (relay) => {496- relay.graspBusy = true;494+ let entered!: () => void;495+ let release!: () => void;496+ const ready = new Promise<void>((resolve) => { entered = resolve; });497+ const hold = new Promise<void>((resolve) => { release = resolve; });498+ const active = relay.repositoryAccess.run("git", async () => {499+ entered();500+ await hold;501+ }, () => { throw new Error("Git scope unexpectedly refused"); });497502 try {503+ await ready;504+ expect(relay.repositoryAccess.busy).toBe(true);505+ expect(relay.repositoryAccess.kind).toBe("git");498506 const result = await relay.acceptAny(event, relay.virtualConn(host, pk(owner)));499507 expect(result.ok).toBe(false);500- expect(result.msg).toContain("Git transaction in progress");508+ expect(result.msg).toContain("relay operation in progress");501509 const response = await relay.fetch(new Request(`http://${host}/`, { method: "POST", body: "{}" }));502510 expect(response.status).toBe(429);503511 await response.text();504512 expect(relay.sql.exec("SELECT 1 FROM events WHERE id=?", event.id).toArray()).toEqual([]);505- } finally { relay.graspBusy = false; }513+ } finally {514+ release();515+ await active;516+ expect(relay.repositoryAccess.busy).toBe(false);517+ }506518 });507519 });508520