diff --git a/src/deploy/functions/release/fabricator.spec.ts b/src/deploy/functions/release/fabricator.spec.ts index bcb17f49f66..bbc417c01de 100644 --- a/src/deploy/functions/release/fabricator.spec.ts +++ b/src/deploy/functions/release/fabricator.spec.ts @@ -343,6 +343,24 @@ describe("Fabricator", () => { await fab.createV1Function(ep1, new scraper.SourceTokenScraper()); expect(gcf.setInvokerCreate).to.not.have.been.called; }); + + it("does not deadlock when createFunction receives a retryable 409 on the first attempt", async () => { + const retryingFab = new fabricator.Fabricator({ + ...ctorArgs, + functionExecutor: new executor.QueueExecutor({ retries: 1, backoff: 10, maxBackoff: 10 }), + }); + + const err409 = new Error("unable to queue the operation"); + (err409 as any).status = 409; + gcf.createFunction.onFirstCall().rejects(err409); + gcf.createFunction.resolves({ name: "op", type: "create", done: false }); + poller.pollOperation.resolves(); + + const ep = endpoint({ scheduleTrigger: {} }); + await expect(retryingFab.createV1Function(ep, new scraper.SourceTokenScraper())).to.eventually + .be.undefined; + expect(gcf.createFunction).to.have.been.calledTwice; + }); }); describe("updateV1Function", () => { @@ -431,6 +449,24 @@ describe("Fabricator", () => { await fab.updateV1Function(ep, new scraper.SourceTokenScraper()); expect(gcf.setInvokerUpdate).to.not.have.been.called; }); + + it("does not deadlock when updateFunction receives a retryable 409 on the first attempt", async () => { + const retryingFab = new fabricator.Fabricator({ + ...ctorArgs, + functionExecutor: new executor.QueueExecutor({ retries: 1, backoff: 10, maxBackoff: 10 }), + }); + + const err409 = new Error("unable to queue the operation"); + (err409 as any).status = 409; + gcf.updateFunction.onFirstCall().rejects(err409); + gcf.updateFunction.resolves({ name: "op", type: "update", done: false }); + poller.pollOperation.resolves(); + + const ep = endpoint({ httpsTrigger: {} }); + await expect(retryingFab.updateV1Function(ep, new scraper.SourceTokenScraper())).to.eventually + .be.undefined; + expect(gcf.updateFunction).to.have.been.calledTwice; + }); }); describe("deleteV1Function", () => { @@ -885,6 +921,29 @@ describe("Fabricator", () => { ); }); + it("does not deadlock when updateFunction receives a retryable 409 on the first attempt", async () => { + // Regression test for: a 409 retry re-entering getToken() while the + // scraper is still FETCHING would await a promise that only the poller + // could resolve — but the poller never ran because the request failed. + // The fix calls scraper.abort() inside the closure catch so the promise + // is settled before the executor's backoff fires and the closure re-runs. + const retryingFab = new fabricator.Fabricator({ + ...ctorArgs, + functionExecutor: new executor.QueueExecutor({ retries: 1, backoff: 10, maxBackoff: 10 }), + }); + + const err409 = new Error("unable to queue the operation"); + (err409 as any).status = 409; + gcfv2.updateFunction.onFirstCall().rejects(err409); + gcfv2.updateFunction.resolves({ name: "op", done: false }); + poller.pollOperation.resolves({ serviceConfig: { service: "service" } }); + + const ep = endpoint({ httpsTrigger: {} }, { platform: "gcfv2" }); + await expect(retryingFab.updateV2Function(ep, new scraper.SourceTokenScraper())).to.eventually + .be.undefined; + expect(gcfv2.updateFunction).to.have.been.calledTwice; + }); + it("throws on set invoker failure", async () => { gcfv2.updateFunction.resolves({ name: "op", done: false }); poller.pollOperation.resolves({ serviceConfig: { service: "service" } }); diff --git a/src/deploy/functions/release/fabricator.ts b/src/deploy/functions/release/fabricator.ts index b43339dd955..da6930f1448 100644 --- a/src/deploy/functions/release/fabricator.ts +++ b/src/deploy/functions/release/fabricator.ts @@ -417,13 +417,18 @@ export class Fabricator { .run(async () => { // try to get the source token right before deploying apiFunction.sourceToken = await scraper.getToken(); - const op: { name: string } = await gcf.createFunction(apiFunction); - return poller.pollOperation({ - ...gcfV1PollerOptions, - pollerName: `create-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, - operationResourceName: op.name, - onPoll: scraper.poller, - }); + try { + const op: { name: string } = await gcf.createFunction(apiFunction); + return poller.pollOperation({ + ...gcfV1PollerOptions, + pollerName: `create-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, + operationResourceName: op.name, + onPoll: scraper.poller, + }); + } catch (err) { + scraper.abort(); + throw err; + } }) .catch(rethrowAs(endpoint, "create")); @@ -546,18 +551,22 @@ export class Fabricator { if (experiments.isEnabled("functionsv2deployoptimizations")) { apiFunction.buildConfig.sourceToken = await scraper.getToken(); } - const op: { name: string } = await gcfV2.createFunction(apiFunction); - return await poller.pollOperation({ - ...gcfV2PollerOptions, - pollerName: `create-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, - operationResourceName: op.name, - onPoll: scraper.poller, - }); + try { + const op: { name: string } = await gcfV2.createFunction(apiFunction); + return await poller.pollOperation({ + ...gcfV2PollerOptions, + pollerName: `create-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, + operationResourceName: op.name, + onPoll: scraper.poller, + }); + } catch (err) { + // Release the scraper so the next retry (or concurrent waiters) are + // not stuck awaiting a token promise that will never be fulfilled. + scraper.abort(); + throw err; + } }) .catch(async (err: any) => { - // Abort waiting on source token so other concurrent calls don't get stuck - scraper.abort(); - // If the createFunction call returns RPC error code RESOURCE_EXHAUSTED (8), // we have exhausted the underlying Cloud Run API quota. To retry, we need to // first delete the GCF function resource, then call createFunction again. @@ -644,13 +653,18 @@ export class Fabricator { const resultFunction = await this.functionExecutor .run(async () => { apiFunction.sourceToken = await scraper.getToken(); - const op: { name: string } = await gcf.updateFunction(apiFunction); - return await poller.pollOperation({ - ...gcfV1PollerOptions, - pollerName: `update-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, - operationResourceName: op.name, - onPoll: scraper.poller, - }); + try { + const op: { name: string } = await gcf.updateFunction(apiFunction); + return await poller.pollOperation({ + ...gcfV1PollerOptions, + pollerName: `update-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, + operationResourceName: op.name, + onPoll: scraper.poller, + }); + } catch (err) { + scraper.abort(); + throw err; + } }) .catch(rethrowAs(endpoint, "update")); @@ -695,18 +709,24 @@ export class Fabricator { if (experiments.isEnabled("functionsv2deployoptimizations")) { apiFunction.buildConfig.sourceToken = await scraper.getToken(); } - const op: { name: string } = await gcfV2.updateFunction(apiFunction); - return await poller.pollOperation({ - ...gcfV2PollerOptions, - pollerName: `update-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, - operationResourceName: op.name, - onPoll: scraper.poller, - }); + try { + const op: { name: string } = await gcfV2.updateFunction(apiFunction); + return await poller.pollOperation({ + ...gcfV2PollerOptions, + pollerName: `update-${endpoint.codebase}-${endpoint.region}-${endpoint.id}`, + operationResourceName: op.name, + onPoll: scraper.poller, + }); + } catch (err) { + // Release the scraper so the next retry (or concurrent waiters) are + // not stuck awaiting a token promise that will never be fulfilled. + scraper.abort(); + throw err; + } }, { retryCodes: [...DEFAULT_RETRY_CODES, CLOUD_RUN_RESOURCE_EXHAUSTED_CODE] }, ) .catch((err: any) => { - scraper.abort(); logger.error((err as Error).message); throw new reporter.DeploymentError(endpoint, "update", err); }); diff --git a/src/deploy/functions/release/sourceTokenScraper.spec.ts b/src/deploy/functions/release/sourceTokenScraper.spec.ts index e3cd4d329f1..aad641ea322 100644 --- a/src/deploy/functions/release/sourceTokenScraper.spec.ts +++ b/src/deploy/functions/release/sourceTokenScraper.spec.ts @@ -80,6 +80,32 @@ describe("SourceTokenScraper", () => { await expect(scraper.getToken()).to.eventually.equal("magic token"); }); + it("abort before a concurrent waiter re-arms the scraper so the retry can proceed", async () => { + const scraper = new SourceTokenScraper(); + // First caller transitions NONE → FETCHING. + await expect(scraper.getToken()).to.eventually.be.undefined; + + // Simulate a failed deploy: abort() is called before the retry fires. + scraper.abort(); + + // The retry's getToken() should resolve immediately (promise already + // settled with {aborted: true}) and return undefined, not deadlock. + const secondToken = scraper.getToken(); + const timeout = new Promise((_, reject) => + setTimeout(() => reject(new Error("deadlock: getToken() did not resolve")), 50), + ); + await expect(Promise.race([secondToken, timeout])).to.eventually.be.undefined; + + // After the re-arm, a subsequent poller call delivers the token normally. + scraper.poller({ + metadata: { + sourceToken: "retry token", + target: "projects/p/locations/l/functions/f", + }, + }); + await expect(scraper.getToken()).to.eventually.equal("retry token"); + }); + it("concurrent requests for source token", async () => { const scraper = new SourceTokenScraper(); diff --git a/src/deploy/functions/release/sourceTokenScraper.ts b/src/deploy/functions/release/sourceTokenScraper.ts index 6cdfc1c368d..614e8002b26 100644 --- a/src/deploy/functions/release/sourceTokenScraper.ts +++ b/src/deploy/functions/release/sourceTokenScraper.ts @@ -1,11 +1,17 @@ import { FirebaseError } from "../../../error"; import { assertExhaustive } from "../../../functional"; import { logger } from "../../../logger"; +import { timeoutFallback } from "../../../timeout"; + +// How long a concurrent getToken() call will wait for the first deploy's poller +// to produce a token before giving up and proceeding without one. +const SOURCE_TOKEN_FETCH_TIMEOUT_MS = 25 * 60 * 1000; // 25 minutes, matches fabricator.ts masterTimeout type TokenFetchState = "NONE" | "FETCHING" | "VALID"; interface TokenFetchResult { token?: string; aborted: boolean; + timedOut?: boolean; } /** @@ -35,8 +41,20 @@ export class SourceTokenScraper { this.fetchState = "FETCHING"; return undefined; } else if (this.fetchState === "FETCHING") { - const tokenResult = await this.promise; + // Race the token promise against a deadline so a lost token producer + // (e.g. a failed deploy that didn't call abort()) degrades to a + // token-less deploy rather than hanging the process forever. + const tokenResult: TokenFetchResult = await timeoutFallback( + this.promise, + { aborted: true, timedOut: true }, + SOURCE_TOKEN_FETCH_TIMEOUT_MS, + ); if (tokenResult.aborted) { + if (tokenResult.timedOut) { + logger.warn( + "Timed out waiting for a source token. Proceeding without one, which may slow the deploy.", + ); + } this.promise = new Promise((resolve) => (this.resolve = resolve)); return undefined; } diff --git a/src/timeout.ts b/src/timeout.ts index acf3e172c68..41b312c4038 100644 --- a/src/timeout.ts +++ b/src/timeout.ts @@ -6,10 +6,17 @@ export async function timeoutFallback( value: V, timeoutMillis = 2000, ): Promise { - return Promise.race([ - promise, - new Promise((resolve) => setTimeout(() => resolve(value), timeoutMillis)), - ]); + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((resolve) => { + timer = setTimeout(() => resolve(value), timeoutMillis); + }), + ]); + } finally { + clearTimeout(timer); + } } export async function timeoutError(