Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,15 @@ import { ManagedSignerService } from "@src/billing/services/managed-signer/manag
import { type ApiPgDatabase, type EnqueueOptions, JOB_NAME, POSTGRES_DB, resolveTable } from "@src/core";
import { FeatureFlags } from "@src/core/services/feature-flags/feature-flags";
import { FeatureFlagsService } from "@src/core/services/feature-flags/feature-flags.service";
import { LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE } from "@src/deployment/services/lease/lease.service";
import { ACCESS_REVOKED_ERROR_CODE } from "@src/deployment/services/member-deployment-access/member-deployment-access.service";
import { pendingLeaseActivityOf } from "@src/deployment/utils/lease-activity/lease-activity";
import { ProviderService } from "@src/provider/services/provider/provider.service";
import { CreateLeaseHandler } from "./create-lease.handler";
import { CREATE_LEASE_RETRY_OPTIONS, CreateLease, createLeaseKeyFor, leaseWalletRefOf } from "./create-lease.job";

import { createAkashAddress } from "@test/seeders/akash-address.seeder";
import { createBid } from "@test/seeders/bid.seeder";
import { createDseq, seedDeploymentSetting } from "@test/seeders/db/deployment-setting.seeder";
import { seedOrganizationMember, seedOrganizationWithOwner, seedProject, seedProjectMember } from "@test/seeders/db/organization.seeder";
import { seedOrganizationWallet, seedUser, seedUserWithWallet } from "@test/seeders/db/user-with-wallet.seeder";
Expand Down Expand Up @@ -88,7 +90,7 @@ describe(CreateLeaseHandler.name, () => {
const { createInBackground, findActivity, awaitCompleted, signerService, providerService } = await setup({
inTeam: true,
revoke: "membership",
landed: true
leaseOnChain: "active"
});

await createInBackground();
Expand Down Expand Up @@ -124,6 +126,27 @@ describe(CreateLeaseHandler.name, () => {
expect(await findActivity()).toMatchObject({ status: "failed", meta: { error: { code: "manifest_not_delivered" } } });
});

it("records that a lease closed before its provider received the manifest, rather than blaming its closed bid", async () => {
const { createInBackground, findActivity, awaitCompleted, bidHttpService, wallet, leases } = await setup({ leaseOnChain: "closed" });
const closedBid = createBid({ owner: wallet.address!, ...leases[0] });
closedBid.bid.state = "closed";
bidHttpService.list.mockResolvedValue([closedBid]);

await createInBackground();

const job = await awaitCompleted();
expect(job.retry_count).toBe(0);
expect(await findActivity()).toMatchObject({
status: "failed",
meta: {
error: {
code: LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE,
message: `The lease for deployment ${leases[0].dseq} was created, but it closed before provider ${leases[0].provider} received its manifest. Close the deployment and deploy again.`
}
}
});
});

it("keeps the activity pending while the lease still has retries to come", async () => {
const { createInBackground, findActivity, waitForJobIn, providerService } = await setup();
providerService.sendManifest.mockRejectedValue(createError(503, "Provider service is temporarily unavailable"));
Expand All @@ -134,7 +157,7 @@ describe(CreateLeaseHandler.name, () => {
expect(await findActivity()).toMatchObject({ status: "pending" });
});

async function setup(input: { inTeam?: boolean; revoke?: Revocation; landed?: boolean } = {}) {
async function setup(input: { inTeam?: boolean; revoke?: Revocation; leaseOnChain?: "active" | "closed" } = {}) {
vi.spyOn(container.resolve(FeatureFlagsService), "isEnabled").mockImplementation(flag => flag !== FeatureFlags.ORGANIZATIONS_ENFORCE);
const { enqueue, startWorkers } = await jobWorkers();
const activityRepository = container.resolve(ActivityRepository);
Expand All @@ -144,12 +167,13 @@ describe(CreateLeaseHandler.name, () => {
const leases = [{ dseq, gseq: 1, oseq: 1, provider: createAkashAddress() }];
const activity = await activityRepository.create(pendingLeaseActivityOf({ userId: actorId, leases }));
const singletonKey = createLeaseKeyFor({ owner: wallet.address!, dseq });
const onChain = input.landed
? [createLeaseApiResponse({ owner: wallet.address!, dseq, gseq: 1, oseq: 1, provider: leases[0].provider, state: "active" })]
const onChain = input.leaseOnChain
? [createLeaseApiResponse({ owner: wallet.address!, dseq, gseq: 1, oseq: 1, provider: leases[0].provider, state: input.leaseOnChain })]
: [];

vi.spyOn(container.resolve(LeaseHttpService), "list").mockResolvedValue({ leases: onChain, pagination: { next_key: null, total: String(onChain.length) } });
vi.spyOn(container.resolve(BidHttpService), "list").mockResolvedValue([]);
const bidHttpService = container.resolve(BidHttpService);
vi.spyOn(bidHttpService, "list").mockResolvedValue([]);
const signerService = container.resolve(ManagedSignerService);
vi.spyOn(signerService, "assertCanBroadcastByUserWallet").mockResolvedValue(undefined);
vi.spyOn(signerService, "executeDecodedTxByUserWallet").mockResolvedValue({ code: 0, hash: "ABCDEF", transactionHash: "ABCDEF", rawLog: "" });
Expand All @@ -162,6 +186,8 @@ describe(CreateLeaseHandler.name, () => {
return {
wallet,
actorId,
leases,
bidHttpService: vi.mocked(bidHttpService),
signerService: vi.mocked(signerService),
providerService: vi.mocked(providerService),
awaitCompleted: () => expectJobCompleted(CreateLease[JOB_NAME], { singletonKey }),
Expand Down
83 changes: 82 additions & 1 deletion apps/api/src/deployment/services/lease/lease.service.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import { ACCESS_REVOKED_ERROR_CODE } from "@src/deployment/services/member-deplo
import { ReconcileLeaseCreate, reconcileLeaseCreateOptionsFor } from "@src/deployment/services/reconcile-lease-create/reconcile-lease-create.job";
import { createdLeaseActivityOf, failedLeaseActivityOf, pendingLeaseActivityOf } from "@src/deployment/utils/lease-activity/lease-activity";
import type { ProviderService } from "@src/provider/services/provider/provider.service";
import { LeaseService } from "./lease.service";
import { LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE, LeaseService } from "./lease.service";

import { createAkashAddress } from "@test/seeders/akash-address.seeder";
import { createBid } from "@test/seeders/bid.seeder";
Expand Down Expand Up @@ -1106,6 +1106,87 @@ describe(LeaseService.name, () => {
expect(activityService.open).not.toHaveBeenCalled();
expect(jobQueueService.enqueue).not.toHaveBeenCalled();
});

describe("when the provider closed a lease an earlier attempt created", () => {
it.each([
{ failure: "the bid check refuses its closed bid", closedBid: true, broadcast: undefined },
{ failure: "the chain refuses the broadcast", closedBid: false, broadcast: LEASE_ON_CLOSED_BID }
])("reports the lease closed before its provider received the manifest when $failure", async ({ closedBid, broadcast }) => {
const lease = { dseq: "100", gseq: 1, oseq: 1, provider: createAkashAddress() };
const { service, signerService, providerService, wallet } = setup({
onChain: owner => [leaseOnChain(owner, lease, "closed")],
bids: owner => (closedBid ? [bidFor(owner, lease, "closed")] : [])
});
if (broadcast) signerService.executeDecodedTxByUserWallet.mockRejectedValue(broadcast);

await expect(service.createQueuedLeases({ wallet, userId: wallet.userId, assertMayStillSign }, [lease])).rejects.toMatchObject({
status: 409,
errorCode: LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE,
message: `The lease for deployment 100 was created, but it closed before provider ${lease.provider} received its manifest. Close the deployment and deploy again.`,
data: { dseq: "100", provider: lease.provider },
originalError: LEASE_ON_CLOSED_BID
});
expect(providerService.sendManifest).not.toHaveBeenCalled();
});

it("names the placement whose lease closed", async () => {
const pending = { dseq: "100", gseq: 1, oseq: 1, provider: createAkashAddress() };
const closed = { dseq: "100", gseq: 2, oseq: 1, provider: createAkashAddress() };
const { service, wallet } = setup({
onChain: owner => [leaseOnChain(owner, closed, "closed")],
bids: owner => [bidFor(owner, closed, "closed")]
});

await expect(service.createQueuedLeases({ wallet, userId: wallet.userId, assertMayStillSign }, [pending, closed])).rejects.toMatchObject({
errorCode: LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE,
data: { dseq: "100", provider: closed.provider }
});
});

it.each([
{ chain: "shows no lease on the placement", onChain: [] },
{ chain: "shows a closed lease only on another group", onChain: [{ dseq: "100", gseq: 2, oseq: 1 }] }
])("keeps the refusal when the chain $chain", async ({ onChain }) => {
const lease = { dseq: "100", gseq: 1, oseq: 1, provider: createAkashAddress() };
const { service, wallet } = setup({
onChain: owner => onChain.map(placement => leaseOnChain(owner, { ...placement, provider: lease.provider }, "closed")),
bids: owner => [bidFor(owner, lease, "closed")]
});

await expect(service.createQueuedLeases({ wallet, userId: wallet.userId, assertMayStillSign }, [lease])).rejects.toBe(LEASE_ON_CLOSED_BID);
});

it("keeps the refusal of a member who lost access, so it stays final", async () => {
const lease = { dseq: "100", gseq: 1, oseq: 1, provider: createAkashAddress() };
const { service, wallet } = setup({ onChain: owner => [leaseOnChain(owner, lease, "closed")] });
const lostAccess = vi.fn(async () => {
throw ACCESS_REVOKED;
});

await expect(service.createQueuedLeases({ wallet, userId: wallet.userId, assertMayStillSign: lostAccess }, [lease])).rejects.toBe(ACCESS_REVOKED);
});

it("keeps the failure of following up a lease that is still live", async () => {
const lease = { dseq: "100", gseq: 1, oseq: 1, provider: createAkashAddress() };
const { service, signerService, wallet } = setup({ onChain: owner => [leaseOnChain(owner, lease)] });
const followUpFailure = createError(503, "Follow-up failed");
signerService.followUpLandedLeases.mockRejectedValue(followUpFailure);

await expect(service.createQueuedLeases({ wallet, userId: wallet.userId, assertMayStillSign }, [lease])).rejects.toBe(followUpFailure);
});

it("keeps the refusal, and logs why, when the leases cannot be read for a closed one", async () => {
const lease = { dseq: "100", gseq: 1, oseq: 1, provider: createAkashAddress() };
const { service, leaseHttpService, logger, wallet } = setup({ bids: owner => [bidFor(owner, lease, "closed")] });
const readFailure = new Error("rpc unavailable");
const nothingOnChain = { leases: [], pagination: { next_key: null, total: "0" } };
leaseHttpService.list.mockResolvedValueOnce(nothingOnChain).mockResolvedValueOnce(nothingOnChain).mockRejectedValueOnce(readFailure);

await expect(service.createQueuedLeases({ wallet, userId: wallet.userId, assertMayStillSign }, [lease])).rejects.toBe(LEASE_ON_CLOSED_BID);

expect(logger.warn).toHaveBeenCalledWith({ event: "CLOSED_LEASE_LOOKUP_FAILED", dseq: "100", error: readFailure });
});
});
});

type TestLease = { dseq: string; gseq: number; oseq: number; provider: string };
Expand Down
46 changes: 42 additions & 4 deletions apps/api/src/deployment/services/lease/lease.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import { type DeploymentResponse } from "@src/deployment/http-schemas/deployment
import { type CreateLeaseRequest } from "@src/deployment/http-schemas/lease.schema";
import { CREATE_LEASE_RETRY_OPTIONS, CreateLease, createLeaseKeyFor, leaseWalletRefOf } from "@src/deployment/services/create-lease/create-lease.job";
import { LeaseManifestService } from "@src/deployment/services/lease-manifest/lease-manifest.service";
import { isAccessRevoked } from "@src/deployment/services/member-deployment-access/member-deployment-access.service";
import { ReconcileLeaseCreate, reconcileLeaseCreateOptionsFor } from "@src/deployment/services/reconcile-lease-create/reconcile-lease-create.job";
import { createdLeaseActivityOf, failedLeaseActivityOf, pendingLeaseActivityOf } from "@src/deployment/utils/lease-activity/lease-activity";
import { isLeaseOn, isSamePlacementSet, type LeasePlacement } from "@src/deployment/utils/lease-placement/lease-placement";
Expand All @@ -24,6 +25,8 @@ import { DeploymentReaderService } from "../deployment-reader/deployment-reader.

type Leases = CreateLeaseRequest["leases"];

export const LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE = "lease_closed_before_manifest";

/** The wallet a lease is signed with and the member acting through it, who is not the wallet's user on a team wallet. */
export type LeaseSigner = { wallet: WalletInitialized; userId: string };

Expand Down Expand Up @@ -89,19 +92,54 @@ export class LeaseService {
@Trace()
public async createQueuedLeases(signer: QueuedLeaseSigner, leases: LeasePlacement[]): Promise<void> {
const manifests = await this.#manifestsByDseq(leases, signer, undefined);
const creation = await this.#planCreation(signer.wallet, leases, signer.assertMayStillSign);

await this.#createOrAdoptLeases(signer, leases, creation);
try {
const creation = await this.#planCreation(signer.wallet, leases, signer.assertMayStillSign);
await this.#createOrAdoptLeases(signer, leases, creation);
} catch (error) {
await this.#rethrowAsClosedBeforeManifestIfLeaseClosed(signer.wallet.address, leases, error);
}

await this.#sendManifests(signer.wallet, leases, manifests);
}

/** Matches each lease on its own placement and provider, since another live lease on the deployment says nothing about the one requested. */
public async leasesExist(owner: string, leases: LeasePlacement[]): Promise<boolean> {
const live = (await this.#listLeases(owner, leases)).filter(lease => lease.state !== "closed");

return leases.every(placement => live.some(lease => isLeaseOn(lease.id, placement)));
}

async #listLeases(owner: string, leases: LeasePlacement[]) {
const dseqs = [...new Set(leases.map(lease => lease.dseq))];
const listed = await Promise.all(dseqs.map(dseq => this.leaseHttpService.list({ owner, dseq })));
const live = listed.flatMap(response => response.leases).filter(({ lease }) => lease.state !== "closed");

return leases.every(placement => live.some(({ lease }) => isLeaseOn(lease.id, placement)));
return listed.flatMap(response => response.leases).map(({ lease }) => lease);
}

/** The request refused closed bids before queueing, so a lease the chain now shows closed was created by an earlier attempt, while a member who lost access keeps that refusal. */
async #rethrowAsClosedBeforeManifestIfLeaseClosed(owner: string, leases: LeasePlacement[], error: unknown): Promise<never> {
if (isAccessRevoked(error)) throw error;

const closed = await this.#placementWithClosedLease(owner, leases);
if (!closed) throw error;

throw createError(
409,
`The lease for deployment ${closed.dseq} was created, but it closed before provider ${closed.provider} received its manifest. Close the deployment and deploy again.`,
{ errorCode: LEASE_CLOSED_BEFORE_MANIFEST_ERROR_CODE, data: { dseq: closed.dseq, provider: closed.provider }, originalError: error }
);
}

/** A read that fails is no sign the lease closed, so the failure that led here stays the answer. */
async #placementWithClosedLease(owner: string, leases: LeasePlacement[]): Promise<LeasePlacement | undefined> {
try {
const closed = (await this.#listLeases(owner, leases)).filter(lease => lease.state === "closed");
return leases.find(placement => closed.some(lease => isLeaseOn(lease.id, placement)));
Comment thread
coderabbitai[bot] marked this conversation as resolved.
} catch (error) {
this.#logger.warn({ event: "CLOSED_LEASE_LOOKUP_FAILED", dseq: leases[0].dseq, error });
return undefined;
}
}

async #queueLeases({ wallet, userId }: LeaseSigner, leases: Leases): Promise<{ activityId: string }> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,16 @@ describe("activityDestinations", () => {
expect(activityDestinationOf(activity)).toEqual({ href: "/deployments/1234", label: "View deployment" });
});

it("leads a deploy whose provider closed its lease before receiving it to its deployment, which no provider can bid on again", () => {
const activity = buildActivity({
type: "lease_create",
status: "failed",
meta: { dseq: "1234", error: { code: "lease_closed_before_manifest", message: "Closed before it was sent." } }
});

expect(activityDestinationOf(activity)).toEqual({ href: "/deployments/1234", label: "View deployment" });
});

it("leads a create about no particular deployment to the deployments list", () => {
const activity = buildActivity({ type: "deployment_create", status: "failed", meta: {} });

Expand Down Expand Up @@ -164,7 +174,11 @@ describe("activityDestinations", () => {
const activity = buildActivity({
type: "wallet_balance_alert",
status: "info",
meta: { alertId: "4f1c2a8e-0b5d-4e6f-9a7b-1c2d3e4f5a6b", summary: "Wallet balance is below 10 AKT", organizationId: "9d3b8f0e-2c4a-4b1e-8f6d-7a5c3e1b9d2f" }
meta: {
alertId: "4f1c2a8e-0b5d-4e6f-9a7b-1c2d3e4f5a6b",
summary: "Wallet balance is below 10 AKT",
organizationId: "9d3b8f0e-2c4a-4b1e-8f6d-7a5c3e1b9d2f"
}
});

expect(activityDestinationOf(activity)).toEqual({ href: "/billing?org=9d3b8f0e-2c4a-4b1e-8f6d-7a5c3e1b9d2f", label: "View billing" });
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { MANIFEST_NOT_DELIVERED } from "@src/hooks/useDeploymentStatus/useDeploymentStatus";
import { deploymentRecoveryOf } from "@src/hooks/useDeploymentStatus/useDeploymentStatus";
import { ACTIVE_ORGANIZATION_QUERY_PARAM } from "@src/lib/active-organization/active-organization-deep-link";
import type { Activity } from "@src/queries/useLatestActivitiesQuery";
import { appendSearchParams, UrlService } from "@src/utils/urlUtils";
Expand Down Expand Up @@ -50,7 +50,8 @@ const DESTINATIONS: Record<Activity["type"], (activity: Activity) => ActivityDes
},
lease_create: ({ status, meta: { dseq, error } }) => {
if (!dseq) return DEPLOYMENT_LIST;
if (status === "failed" && error?.code !== MANIFEST_NOT_DELIVERED) return { href: UrlService.configureDeployment({ dseq }), label: "Choose a provider" };
if (status === "failed" && deploymentRecoveryOf(error?.code) === "choose-provider")
return { href: UrlService.configureDeployment({ dseq }), label: "Choose a provider" };

return { href: UrlService.deploymentDetails(dseq), label: "View deployment" };
},
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,10 @@ describe("DeploymentDetail", () => {

expect(router.replace).not.toHaveBeenCalled();
expect(DeploymentLaunchState).toHaveBeenCalledWith(
expect.objectContaining({ launch: { phase: "failed", providers: [], hasLease: false, reason: "The lease was not created." }, canManage: true }),
expect.objectContaining({
launch: { phase: "failed", providers: [], recovery: "choose-provider", reason: "The lease was not created." },
canManage: true
}),
expect.anything()
);
expect(screen.queryByText("placements")).not.toBeInTheDocument();
Expand All @@ -340,7 +343,7 @@ describe("DeploymentDetail", () => {
});

expect(DeploymentLaunchState).toHaveBeenCalledWith(
expect.objectContaining({ leases, launch: expect.objectContaining({ hasLease: true }) }),
expect.objectContaining({ leases, launch: expect.objectContaining({ recovery: "send-again" }) }),
expect.anything()
);
expect(screen.getByText("placements")).toBeInTheDocument();
Expand Down
Loading
Loading