diff --git a/.changeset/resource-operation-outcomes.md b/.changeset/resource-operation-outcomes.md new file mode 100644 index 0000000..9317acb --- /dev/null +++ b/.changeset/resource-operation-outcomes.md @@ -0,0 +1,10 @@ +--- +"@notation/aws.iac": minor +"@notation/cli": minor +"@notation/core": minor +"@notation/reconciler": minor +"@notation/resource": minor +"@notation/std.iac": minor +--- + +A resource `read` now returns the remote object or throws the tagged `ResourceNotFoundError`. Operations that have started but not settled throw `ResourceOperationPendingError` with their retry delay and optional callback context. The reconciler follows those explicit instructions instead of guessing retry behaviour from provider errors or call-site context. diff --git a/docs/internals/reconciler.md b/docs/internals/reconciler.md index 906273d..6b0d413 100644 --- a/docs/internals/reconciler.md +++ b/docs/internals/reconciler.md @@ -81,18 +81,26 @@ Each CRUD operation is implemented as an async generator with retry support: - **`deleteResourceOperation`** – deletes the resource, removes the entry from state backend - **`readResourceOperation`** – reads current state from the cloud provider (used for drift detection) -### Retry and polling +### Pending operations -Operations support polling for eventual consistency: +A resource operation throws `ResourceOperationPendingError` when it has not finished. The reconciler reads two fields from the error: + +| Field | Action | +| ----- | ------ | +| `retryAfterMs` | Wait this many milliseconds. | +| `callbackContext` | Pass this value to the next call of the same operation. | + +The reconciler then calls the same operation again. Any other error fails the operation. See [Operation errors](./resource.md#operation-errors) for the complete API. + +The default limit is 30 calls to one operation: ```ts [packages/reconciler/src/index.ts] { - maxAttempts: 10, - retryInterval: 2000, + maxOperationAttempts: 30, } ``` -This handles AWS services that return success before the resource is fully available. For example, after creating an IAM Role, a Lambda function may briefly fail to deploy until the role propagates. The retry loop handles cases like this. +The last pending error becomes a failure when the limit is reached. ### Operation lifecycle diff --git a/docs/internals/resource.md b/docs/internals/resource.md index 4cf0921..baae261 100644 --- a/docs/internals/resource.md +++ b/docs/internals/resource.md @@ -137,19 +137,79 @@ All schema items carry these fields: ## Operations -`defineOperations` accepts CRUD handlers and error-handling configuration: - -| Field | Required | Signature / Description | -| ---------------------- | -------- | ------------------------------------------------------------------------------------------------------- | -| `create` | yes | `(params: Params) => Promise>` – create the resource, return its computed key. | -| `read` | no | `(key: CompoundKey) => Promise>` – read current state. | -| `update` | no | `(key, patch, params, state) => Promise` – apply a partial update. | -| `delete` | yes | `(key, state) => Promise` – destroy the resource. | -| `deriveParams` | no | Computes intrinsic derived params from config (not dependency-aware). | -| `retryReadOnCondition` | no | Conditions on read output that trigger a retry (e.g. eventual consistency). | -| `failOnError` | no | Error matchers that cause immediate failure with a reason. | -| `notFoundOnError` | no | Error matchers that indicate the resource does not exist. | -| `retryLaterOnError` | no | Error matchers that indicate a transient failure worth retrying. | +`defineOperations` accepts CRUD handlers and parameter derivation: + +| Field | Required | Signature / Description | +| -------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------- | +| `create` | yes | `(params, context?) => Promise>` – create the resource, return its computed key. | +| `read` | no | `(key, context?) => Promise>` – return the remote object or throw `ResourceNotFoundError`. | +| `update` | no | `(key, patch, params, state, context?) => Promise` – apply a partial update. | +| `delete` | yes | `(key, state, context?) => Promise` – ensure the resource is absent. Implementations must also succeed when the remote resource is already gone. | +| `deriveParams` | no | Computes intrinsic derived params from config (not dependency-aware). | + +### Operation errors + +Import the errors from `@notation/resource`: + +```ts +import { + ResourceNotFoundError, + ResourceOperationPendingError, +} from "@notation/resource"; +``` + +The constructors are: + +```ts +new ResourceNotFoundError(message: string, options?: { cause?: unknown }); + +new ResourceOperationPendingError(message: string, { + retryAfterMs: number; + callbackContext?: Readonly>; + cause?: unknown; +}); +``` + +| Handler result | Meaning | What the reconciler does | +| -------------- | ------- | ------------------------ | +| Return normally | The operation finished. | Continues the deployment. | +| `throw new ResourceNotFoundError(message, { cause })` | `read` found no resource for the given key. | Treats the resource as absent during planning and refresh. A read after create or update fails because that operation claimed to have finished. | +| `throw new ResourceOperationPendingError(message, { retryAfterMs, callbackContext })` | The operation has not finished. | Waits for `retryAfterMs`, then calls the same handler again. It passes `callbackContext` as the handler's final argument. | +| Throw any other error | The operation failed. | Stops the deployment. | + +`ResourceNotFoundError` is for `read`. A `delete` handler must catch the provider's missing-resource error and return normally. + +`ResourceOperationPendingError` may be thrown by `create`, `read`, `update`, or `delete`. Its options are: + +| Option | Type | Required | Meaning | +| ------ | ---- | -------- | ------- | +| `retryAfterMs` | `number` | yes | Milliseconds to wait. It must be zero or greater. | +| `callbackContext` | `Readonly>` | no | Plain serializable data for the next attempt. | +| `cause` | `unknown` | no | The provider error that caused this result. | + +The default limit is 30 attempts. Set `maxOperationAttempts` on the reconciler to change it. Reaching the limit fails the operation. + +```ts +read: async (key, context) => { + try { + return await client.send(new GetResourceCommand(key)); + } catch (error) { + if (error instanceof ResourceMissingException) { + throw new ResourceNotFoundError("Resource was not found", { + cause: error, + }); + } + if (error instanceof OperationInProgressException) { + throw new ResourceOperationPendingError("Resource is not ready", { + retryAfterMs: 1_000, + callbackContext: { requestId: error.requestId }, + cause: error, + }); + } + throw error; + } +}; +``` ## Dependencies diff --git a/examples/reconciler/package.json b/examples/reconciler/package.json index 754b89a..55cbaa1 100644 --- a/examples/reconciler/package.json +++ b/examples/reconciler/package.json @@ -11,7 +11,8 @@ "dependencies": { "@notation/reconciler": "workspace:*", "@notation/resource": "workspace:*", - "@notation/state-sqlite": "workspace:*" + "@notation/state-sqlite": "workspace:*", + "@notation/utils": "workspace:*" }, "devDependencies": { "@types/node": "^22.13.4", diff --git a/examples/reconciler/src/static-site.ts b/examples/reconciler/src/static-site.ts index 297a00c..6e4eacf 100644 --- a/examples/reconciler/src/static-site.ts +++ b/examples/reconciler/src/static-site.ts @@ -1,6 +1,7 @@ import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; import path from "node:path"; -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; +import { isErrorWithCode } from "@notation/utils"; type StaticSiteApi = { Key: { siteDirectory: string }; @@ -9,10 +10,6 @@ type StaticSiteApi = { ReadResult: { html: string }; }; -class SiteNotFound extends Error { - readonly name = "SiteNotFound"; -} - const staticSite = resource({ type: "local/site/static" }); export const StaticSite = staticSite @@ -40,7 +37,11 @@ export const StaticSite = staticSite ); return { html }; } catch (error) { - if (isFileMissing(error)) throw new SiteNotFound(siteDirectory); + if (isErrorWithCode(error, "ENOENT")) { + throw new ResourceNotFoundError("Static site was not found", { + cause: error, + }); + } throw error; } }, @@ -48,16 +49,6 @@ export const StaticSite = staticSite await writeFile(path.join(siteDirectory, "index.html"), html, "utf8"); }, delete: async ({ siteDirectory }) => { - await rm(siteDirectory, { recursive: true }); + await rm(siteDirectory, { recursive: true, force: true }); }, - notFoundOnError: [{ name: "SiteNotFound", reason: "site was removed" }], }); - -function isFileMissing(error: unknown): boolean { - return ( - typeof error === "object" && - error !== null && - "code" in error && - error.code === "ENOENT" - ); -} diff --git a/packages/aws.iac/src/resources/api-gateway/api.ts b/packages/aws.iac/src/resources/api-gateway/api.ts index 712cc4e..0e95ed7 100644 --- a/packages/aws.iac/src/resources/api-gateway/api.ts +++ b/packages/aws.iac/src/resources/api-gateway/api.ts @@ -1,4 +1,4 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import * as sdk from "@aws-sdk/client-apigatewayv2"; import { apiGatewayClient } from "src/utils/aws-clients"; import { AwsSchema } from "src/utils/types"; @@ -97,19 +97,32 @@ export const Api = apiSchema.defineOperations({ return { ApiId: result.ApiId! }; }, async read(key) { - const command = new sdk.GetApiCommand(key); - const result = await apiGatewayClient.send(command); - // todo: check types or correct or if RouteKey is actually in result - // if not, need to pass the original params to read - return { RouteKey: "", ...result }; + try { + const command = new sdk.GetApiCommand(key); + const result = await apiGatewayClient.send(command); + // todo: check types or correct or if RouteKey is actually in result + // if not, need to pass the original params to read + return { RouteKey: "", ...result }; + } catch (error) { + if (error instanceof sdk.NotFoundException) { + throw new ResourceNotFoundError("API Gateway API was not found", { + cause: error, + }); + } + throw error; + } }, async update(key, params) { const command = new sdk.UpdateApiCommand({ ...key, ...params }); await apiGatewayClient.send(command); }, async delete(pk) { - const command = new sdk.DeleteApiCommand(pk); - await apiGatewayClient.send(command); + try { + const command = new sdk.DeleteApiCommand(pk); + await apiGatewayClient.send(command); + } catch (error) { + if (!(error instanceof sdk.NotFoundException)) throw error; + } }, }); diff --git a/packages/aws.iac/src/resources/api-gateway/auth.ts b/packages/aws.iac/src/resources/api-gateway/auth.ts index a56996b..f62e673 100644 --- a/packages/aws.iac/src/resources/api-gateway/auth.ts +++ b/packages/aws.iac/src/resources/api-gateway/auth.ts @@ -1,6 +1,6 @@ import * as sdk from "@aws-sdk/client-apigatewayv2"; import { AwsSchema } from "src/utils/types"; -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import { apiGatewayClient } from "src/utils/aws-clients"; import { ApiInstance } from "./api"; @@ -59,18 +59,31 @@ export const RouteAuth = apiSchema }; }, read: async (key) => { - const command = new sdk.GetAuthorizerCommand(key); - const result = await apiGatewayClient.send(command); - - return result; + try { + const command = new sdk.GetAuthorizerCommand(key); + const output = await apiGatewayClient.send(command); + return output; + } catch (error) { + if (error instanceof sdk.NotFoundException) { + throw new ResourceNotFoundError( + "API Gateway authorizer was not found", + { cause: error }, + ); + } + throw error; + } }, update: async (key, patch, params) => { const command = new sdk.UpdateAuthorizerCommand({ ...key, ...params }); await apiGatewayClient.send(command); }, delete: async (params) => { - const command = new sdk.DeleteAuthorizerCommand(params); - await apiGatewayClient.send(command); + try { + const command = new sdk.DeleteAuthorizerCommand(params); + await apiGatewayClient.send(command); + } catch (error) { + if (!(error instanceof sdk.NotFoundException)) throw error; + } }, }) .requireDependencies() diff --git a/packages/aws.iac/src/resources/api-gateway/lambda-integration.ts b/packages/aws.iac/src/resources/api-gateway/lambda-integration.ts index 9390380..a65f963 100644 --- a/packages/aws.iac/src/resources/api-gateway/lambda-integration.ts +++ b/packages/aws.iac/src/resources/api-gateway/lambda-integration.ts @@ -1,4 +1,4 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import * as sdk from "@aws-sdk/client-apigatewayv2"; import { ApiInstance } from "./api"; import { LambdaFunctionInstance } from "../lambda"; @@ -118,16 +118,31 @@ export const LambdaIntegration = integrationSchema return { IntegrationId: result.IntegrationId! }; }, read: async (key) => { - const command = new sdk.GetIntegrationCommand(key); - return apiGatewayClient.send(command); + try { + const command = new sdk.GetIntegrationCommand(key); + const output = await apiGatewayClient.send(command); + return output; + } catch (error) { + if (error instanceof sdk.NotFoundException) { + throw new ResourceNotFoundError( + "API Gateway integration was not found", + { cause: error }, + ); + } + throw error; + } }, update: async (key, params) => { const command = new sdk.UpdateIntegrationCommand({ ...key, ...params }); await apiGatewayClient.send(command); }, delete: async (key) => { - const command = new sdk.DeleteIntegrationCommand(key); - await apiGatewayClient.send(command); + try { + const command = new sdk.DeleteIntegrationCommand(key); + await apiGatewayClient.send(command); + } catch (error) { + if (!(error instanceof sdk.NotFoundException)) throw error; + } }, }) .requireDependencies() diff --git a/packages/aws.iac/src/resources/api-gateway/route.ts b/packages/aws.iac/src/resources/api-gateway/route.ts index e381f04..f22b9db 100644 --- a/packages/aws.iac/src/resources/api-gateway/route.ts +++ b/packages/aws.iac/src/resources/api-gateway/route.ts @@ -1,4 +1,4 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import * as sdk from "@aws-sdk/client-apigatewayv2"; import { apiGatewayClient } from "src/utils/aws-clients"; import { ApiInstance, LambdaIntegrationInstance } from "."; @@ -92,17 +92,30 @@ export const Route = routeSchema return { RouteId: result.RouteId! }; }, read: async (key) => { - const command = new sdk.GetRouteCommand(key); - const result = await apiGatewayClient.send(command); - return { ...key, ...result }; + try { + const command = new sdk.GetRouteCommand(key); + const result = await apiGatewayClient.send(command); + return { ...key, ...result }; + } catch (error) { + if (error instanceof sdk.NotFoundException) { + throw new ResourceNotFoundError("API Gateway route was not found", { + cause: error, + }); + } + throw error; + } }, update: async (key, patch, params) => { const command = new sdk.UpdateRouteCommand({ ...key, ...params }); await apiGatewayClient.send(command); }, delete: async (key) => { - const command = new sdk.DeleteRouteCommand(key); - await apiGatewayClient.send(command); + try { + const command = new sdk.DeleteRouteCommand(key); + await apiGatewayClient.send(command); + } catch (error) { + if (!(error instanceof sdk.NotFoundException)) throw error; + } }, }) .requireDependencies() diff --git a/packages/aws.iac/src/resources/api-gateway/stage.ts b/packages/aws.iac/src/resources/api-gateway/stage.ts index c253b69..898bac2 100644 --- a/packages/aws.iac/src/resources/api-gateway/stage.ts +++ b/packages/aws.iac/src/resources/api-gateway/stage.ts @@ -1,4 +1,4 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import * as sdk from "@aws-sdk/client-apigatewayv2"; import { ApiInstance } from "./api"; import { apiGatewayClient } from "src/utils/aws-clients"; @@ -76,16 +76,30 @@ export const Stage = stageSchema await apiGatewayClient.send(command); }, read: async (key) => { - const command = new sdk.GetStageCommand(key); - return apiGatewayClient.send(command); + try { + const command = new sdk.GetStageCommand(key); + const output = await apiGatewayClient.send(command); + return output; + } catch (error) { + if (error instanceof sdk.NotFoundException) { + throw new ResourceNotFoundError("API Gateway stage was not found", { + cause: error, + }); + } + throw error; + } }, update: async (key, params) => { const command = new sdk.UpdateStageCommand({ ...key, ...params }); await apiGatewayClient.send(command); }, delete: async (key) => { - const command = new sdk.DeleteStageCommand(key); - await apiGatewayClient.send(command); + try { + const command = new sdk.DeleteStageCommand(key); + await apiGatewayClient.send(command); + } catch (error) { + if (!(error instanceof sdk.NotFoundException)) throw error; + } }, }) .requireDependencies() diff --git a/packages/aws.iac/src/resources/event-bridge/rule.ts b/packages/aws.iac/src/resources/event-bridge/rule.ts index a2d1150..66bf699 100644 --- a/packages/aws.iac/src/resources/event-bridge/rule.ts +++ b/packages/aws.iac/src/resources/event-bridge/rule.ts @@ -1,4 +1,4 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import { AwsSchema } from "src/utils/types"; import * as sdk from "@aws-sdk/client-eventbridge"; import { eventBridgeClient } from "src/utils/aws-clients"; @@ -60,21 +60,31 @@ const eventBridgeRuleSchema = eventBridgeRule.defineSchema({ export const EventBridgeRule = eventBridgeRuleSchema .defineOperations({ read: async (key) => { - const describeRuleCommand = new sdk.DescribeRuleCommand(key); - const listRuleTargetsCommand = new sdk.ListTargetsByRuleCommand({ - Rule: key.Name, - EventBusName: key.EventBusName, - }); - - const [ruleDescriptionResult, listRuleTargetsResult] = await Promise.all([ - eventBridgeClient.send(describeRuleCommand), - eventBridgeClient.send(listRuleTargetsCommand), - ]); + try { + const describeRuleCommand = new sdk.DescribeRuleCommand(key); + const listRuleTargetsCommand = new sdk.ListTargetsByRuleCommand({ + Rule: key.Name, + EventBusName: key.EventBusName, + }); - return { - ...ruleDescriptionResult, - ...listRuleTargetsResult, - }; + const [ruleDescriptionResult, listRuleTargetsResult] = + await Promise.all([ + eventBridgeClient.send(describeRuleCommand), + eventBridgeClient.send(listRuleTargetsCommand), + ]); + + return { + ...ruleDescriptionResult, + ...listRuleTargetsResult, + }; + } catch (error) { + if (error instanceof sdk.ResourceNotFoundException) { + throw new ResourceNotFoundError("EventBridge rule was not found", { + cause: error, + }); + } + throw error; + } }, create: async (params) => { @@ -102,27 +112,30 @@ export const EventBridgeRule = eventBridgeRuleSchema await eventBridgeClient.send(updateTargetsCommand); }, delete: async (key) => { - // The targets must be deleted first, otherwise the API will return an error response - - const existingTargets = await eventBridgeClient.send( - new sdk.ListTargetsByRuleCommand({ - Rule: key.Name, - EventBusName: key.EventBusName, - }), - ); - - if (existingTargets.Targets && existingTargets.Targets.length > 0) { - const deleteTargetsCommand = new sdk.RemoveTargetsCommand({ - Rule: key.Name, - EventBusName: key.EventBusName, - Ids: existingTargets.Targets.map((target) => target.Id!), - }); - - await eventBridgeClient.send(deleteTargetsCommand); + try { + // The targets must be deleted first, otherwise the API will return an error response + const existingTargets = await eventBridgeClient.send( + new sdk.ListTargetsByRuleCommand({ + Rule: key.Name, + EventBusName: key.EventBusName, + }), + ); + + if (existingTargets.Targets && existingTargets.Targets.length > 0) { + const deleteTargetsCommand = new sdk.RemoveTargetsCommand({ + Rule: key.Name, + EventBusName: key.EventBusName, + Ids: existingTargets.Targets.map((target) => target.Id!), + }); + + await eventBridgeClient.send(deleteTargetsCommand); + } + + const deleteRuleCommand = new sdk.DeleteRuleCommand(key); + await eventBridgeClient.send(deleteRuleCommand); + } catch (error) { + if (!(error instanceof sdk.ResourceNotFoundException)) throw error; } - - const deleteRuleCommand = new sdk.DeleteRuleCommand(key); - await eventBridgeClient.send(deleteRuleCommand); }, }) .requireDependencies() diff --git a/packages/aws.iac/src/resources/lambda/lambda-role.ts b/packages/aws.iac/src/resources/lambda/lambda-role.ts index 51c5f39..27933e9 100644 --- a/packages/aws.iac/src/resources/lambda/lambda-role.ts +++ b/packages/aws.iac/src/resources/lambda/lambda-role.ts @@ -1,4 +1,4 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import * as sdk from "@aws-sdk/client-iam"; import { iamClient } from "src/utils/aws-clients"; import { AwsSchema } from "src/utils/types"; @@ -69,17 +69,30 @@ export const LambdaIamRole = lambdaIamRoleSchema.defineOperations({ await iamClient.send(command); }, read: async (key) => { - const command = new sdk.GetRoleCommand(key); - const { Role } = await iamClient.send(command); - return Role!; + try { + const command = new sdk.GetRoleCommand(key); + const { Role } = await iamClient.send(command); + return Role!; + } catch (error) { + if (error instanceof sdk.NoSuchEntityException) { + throw new ResourceNotFoundError("Lambda IAM role was not found", { + cause: error, + }); + } + throw error; + } }, update: async (key, params) => { const command = new sdk.UpdateRoleCommand({ ...key, ...params }); await iamClient.send(command); }, delete: async (key) => { - const command = new sdk.DeleteRoleCommand(key); - await iamClient.send(command); + try { + const command = new sdk.DeleteRoleCommand(key); + await iamClient.send(command); + } catch (error) { + if (!(error instanceof sdk.NoSuchEntityException)) throw error; + } }, deriveParams: () => ({ AssumeRolePolicyDocument: JSON.stringify(lambdaTrustPolicy), diff --git a/packages/aws.iac/src/resources/lambda/lambda.ts b/packages/aws.iac/src/resources/lambda/lambda.ts index 12d75a2..08f759e 100644 --- a/packages/aws.iac/src/resources/lambda/lambda.ts +++ b/packages/aws.iac/src/resources/lambda/lambda.ts @@ -1,4 +1,9 @@ -import { resource, typed } from "@notation/resource"; +import { + resource, + ResourceNotFoundError, + ResourceOperationPendingError, + typed, +} from "@notation/resource"; import * as sdk from "@aws-sdk/client-lambda"; import { lambdaClient } from "src/utils/aws-clients"; import { AwsSchema } from "src/utils/types"; @@ -186,7 +191,7 @@ export const LambdaFunction = lambdaFunctionSchema Code: { ZipFile: params.Code.ZipFile }, }); - await lambdaClient.send(command); + await runLambdaMutation(() => lambdaClient.send(command)); // if (params.ReservedConcurrentExecutions) { // const concurrencyCommand = new sdk.PutFunctionConcurrencyCommand({ @@ -198,21 +203,43 @@ export const LambdaFunction = lambdaFunctionSchema }, read: async (key) => { - const command = new sdk.GetFunctionCommand(key); - const { Code, Configuration, Concurrency } = - await lambdaClient.send(command); + try { + const command = new sdk.GetFunctionCommand(key); + const { Code, Configuration, Concurrency } = + await lambdaClient.send(command); - return { - ...Configuration, - Layers: Configuration!.Layers?.map((layer) => layer.Arn), - ...Concurrency, - Code: { - S3Bucket: Code?.Location?.split("/")[0], - S3Key: Code?.Location?.split("/")[1], - S3ObjectVersion: Code?.Location?.split("/")[2], - ZipFile: undefined, - }, - }; + if (Configuration?.State !== "Active") { + throw new ResourceOperationPendingError( + "Waiting for Lambda to become active", + { retryAfterMs: 1_000 }, + ); + } + if (!Configuration.RevisionId) { + throw new ResourceOperationPendingError( + "Waiting for Lambda to be deployed", + { retryAfterMs: 1_000 }, + ); + } + + return { + ...Configuration, + Layers: Configuration.Layers?.map((layer) => layer.Arn), + ...Concurrency, + Code: { + S3Bucket: Code?.Location?.split("/")[0], + S3Key: Code?.Location?.split("/")[1], + S3ObjectVersion: Code?.Location?.split("/")[2], + ZipFile: undefined, + }, + }; + } catch (error) { + if (error instanceof sdk.ResourceNotFoundException) { + throw new ResourceNotFoundError("Lambda function was not found", { + cause: error, + }); + } + throw error; + } }, update: async (key, patch, params) => { @@ -226,7 +253,7 @@ export const LambdaFunction = lambdaFunctionSchema ...key, ...conf, }); - await lambdaClient.send(confCommand); + await runLambdaMutation(() => lambdaClient.send(confCommand)); } if (CodeSha256) { @@ -234,39 +261,18 @@ export const LambdaFunction = lambdaFunctionSchema ...key, ZipFile: params.Code.ZipFile, }); - await lambdaClient.send(codeCommand); + await runLambdaMutation(() => lambdaClient.send(codeCommand)); } }, delete: async (key) => { - const command = new sdk.DeleteFunctionCommand(key); - await lambdaClient.send(command); + try { + const command = new sdk.DeleteFunctionCommand(key); + await runLambdaMutation(() => lambdaClient.send(command)); + } catch (error) { + if (!(error instanceof sdk.ResourceNotFoundException)) throw error; + } }, - retryLaterOnError: [ - { - name: "InvalidParameterValueException", - message: - "The role defined for the function cannot be assumed by Lambda.", - reason: "Waiting for IAM role to propagate", - }, - { - name: "InvalidParameterValueException", - message: "The provided execution role does not have permissions", - // todo: find real reason this is here - reason: "Waiting for IAM role to propagate", - }, - ], - retryReadOnCondition: [ - { - key: "State", - value: "Active", - reason: "Waiting for lambda to become active", - }, - { - key: "RevisionId", - reason: "Waiting for lambda to be deployed", - }, - ], }) .requireDependencies() .deriveParams(async ({ deps }) => ({ @@ -278,6 +284,34 @@ export const LambdaFunction = lambdaFunctionSchema export type LambdaFunctionInstance = InstanceType; +async function runLambdaMutation(mutation: () => Promise): Promise { + try { + return await mutation(); + } catch (error) { + if (isIamPropagationFailure(error)) { + throw new ResourceOperationPendingError( + "Waiting for IAM role to propagate", + { retryAfterMs: 1_000, cause: error }, + ); + } + throw error; + } +} + +function isIamPropagationFailure( + error: unknown, +): error is sdk.InvalidParameterValueException { + if (!(error instanceof sdk.InvalidParameterValueException)) return false; + return ( + error.message.startsWith( + "The role defined for the function cannot be assumed by Lambda.", + ) || + error.message.startsWith( + "The provided execution role does not have permissions", + ) + ); +} + export type LambdaFunctionConfig = ConstructorParameters< typeof LambdaFunction >[0]["config"]; diff --git a/packages/core/src/orchestrator/resource.ts b/packages/core/src/orchestrator/resource.ts index 58dcf81..6281556 100644 --- a/packages/core/src/orchestrator/resource.ts +++ b/packages/core/src/orchestrator/resource.ts @@ -1,9 +1,6 @@ export type { BaseResource, ResourceType, - ErrorMatcher, - ResultCondition, - ResultConditions, ResourceOpts, DefineResourceMeta, DefineResourceApiSchema, @@ -12,8 +9,16 @@ export type { ResourceClass, ResourceSchemaBuilder, ResourceBuilder, + ResourceOperationContext, + ResourceOperationSignal, Schema, SchemaItem, } from "@notation/resource"; -export { Resource, defineResource, resource } from "@notation/resource"; +export { + Resource, + ResourceNotFoundError, + ResourceOperationPendingError, + defineResource, + resource, +} from "@notation/resource"; diff --git a/packages/core/test/orchestrator/resource.doubles.ts b/packages/core/test/orchestrator/resource.doubles.ts index 5c7f10e..a516ed5 100644 --- a/packages/core/test/orchestrator/resource.doubles.ts +++ b/packages/core/test/orchestrator/resource.doubles.ts @@ -65,7 +65,11 @@ export const testOperations = { async delete() {}, async update() {}, async read() { - return { primaryKey: "", optionalSecondaryKey: "", requiredParam: "" }; + return { + primaryKey: "", + optionalSecondaryKey: "", + requiredParam: "", + } as const; }, deriveParams() { return { intrinsicParam: true }; diff --git a/packages/core/test/provisioner/operation.create.test.ts b/packages/core/test/provisioner/operation.create.test.ts index 2de482d..2e4f506 100644 --- a/packages/core/test/provisioner/operation.create.test.ts +++ b/packages/core/test/provisioner/operation.create.test.ts @@ -1,5 +1,9 @@ import { describe, expect, it, vi } from "vitest"; -import { createResourceOperation, createStepRunner, runOperation } from "@notation/reconciler"; +import { + createResourceOperation, + createStepRunner, + runOperation, +} from "@notation/reconciler"; import { MemoryStateBackend } from "@notation/state"; import { TestResourceSchema, @@ -38,7 +42,7 @@ describe("resource creation", () => { const params = await testResource.getParams(); const persistedOutput = testResource.toState(readResult); - expect(createMock.mock.calls[0]).toEqual([params]); + expect(createMock.mock.calls[0]).toEqual([params, undefined]); await expect(stateBackend.get(testResource.id)).resolves.toMatchObject({ id: testResource.id, output: persistedOutput, diff --git a/packages/reconciler/src/operations/index.ts b/packages/reconciler/src/operations/index.ts index 04c66b6..b0107e7 100644 --- a/packages/reconciler/src/operations/index.ts +++ b/packages/reconciler/src/operations/index.ts @@ -1,4 +1,5 @@ export * from "./operation.types"; +export * from "./operation.pending"; export * from "./operation.create"; export * from "./operation.read"; export * from "./operation.update"; diff --git a/packages/reconciler/src/operations/operation.create.ts b/packages/reconciler/src/operations/operation.create.ts index 98eac89..e0aee4f 100644 --- a/packages/reconciler/src/operations/operation.create.ts +++ b/packages/reconciler/src/operations/operation.create.ts @@ -1,12 +1,11 @@ -import { RetryableError, createWorkflow } from "yieldstar"; +import { createWorkflow } from "yieldstar"; import { - DEFAULT_RETRY_OPTIONS, type CreateResourceParams, type StepRunner, emitLifecycleEvent, getErrorDetails, - matchError, } from "./operation.types"; +import { runPendingOperation } from "./operation.pending"; import { readResourceOperation } from "./operation.read"; export async function* createResourceOperation( @@ -25,19 +24,12 @@ export async function* createResourceOperation( params.resource.getParams(), ); - const computedPrimaryKey = yield* step.run("create:remote", async () => { - try { - return await params.resource.create(resourceParams); - } catch (err) { - const matcher = matchError(err, params.resource.retryLaterOnError); - if (matcher) { - throw new RetryableError(matcher.reason, { - ...(params.retryOptions ?? DEFAULT_RETRY_OPTIONS), - }); - } - throw err; - } - }); + const computedPrimaryKey = yield* runPendingOperation( + step, + "create:remote", + (context) => params.resource.create(resourceParams, context), + params.maxOperationAttempts, + ); params.resource.setOutput(resourceParams); if (computedPrimaryKey) { @@ -51,7 +43,7 @@ export async function* createResourceOperation( resource: params.resource, state: params.state, emit: params.emit, - readPollOptions: params.readPollOptions, + maxOperationAttempts: params.maxOperationAttempts, }); params.resource.setOutput({ diff --git a/packages/reconciler/src/operations/operation.delete.ts b/packages/reconciler/src/operations/operation.delete.ts index ffa5de4..51975b6 100644 --- a/packages/reconciler/src/operations/operation.delete.ts +++ b/packages/reconciler/src/operations/operation.delete.ts @@ -1,12 +1,11 @@ -import { RetryableError, createWorkflow } from "yieldstar"; +import { createWorkflow } from "yieldstar"; import { - DEFAULT_RETRY_OPTIONS, type DeleteResourceParams, type StepRunner, emitLifecycleEvent, getErrorDetails, - matchError, } from "./operation.types"; +import { runPendingOperation } from "./operation.pending"; export async function* deleteResourceOperation( step: StepRunner, @@ -20,33 +19,17 @@ export async function* deleteResourceOperation( } try { - try { - yield* step.run("delete:remote", async () => { - try { - await params.resource.delete( - params.resource.key, - params.resource.toState(params.resource.output), - ); - } catch (err) { - const matcher = matchError(err, params.resource.retryLaterOnError); - if (matcher) { - throw new RetryableError(matcher.reason, { - ...(params.retryOptions ?? DEFAULT_RETRY_OPTIONS), - }); - } - throw err; - } - }); - } catch (err) { - const matcher = matchError(err, params.resource.notFoundOnError); - if (matcher) { - await emitLifecycleEvent(params, "delete", "skip", { - reason: matcher.reason, - }); - } else { - throw err; - } - } + yield* runPendingOperation( + step, + "delete:remote", + (context) => + params.resource.delete( + params.resource.key, + params.resource.toState(params.resource.output), + context, + ), + params.maxOperationAttempts, + ); yield* step.run("delete:persist-state", () => params.state.delete(params.resource.id, params.expectedRev), diff --git a/packages/reconciler/src/operations/operation.pending.ts b/packages/reconciler/src/operations/operation.pending.ts new file mode 100644 index 0000000..04a6b7e --- /dev/null +++ b/packages/reconciler/src/operations/operation.pending.ts @@ -0,0 +1,43 @@ +import { + ResourceOperationPendingError, + type ResourceOperationContext, +} from "@notation/resource"; +import type { StepRunner } from "./operation.types"; + +export const DEFAULT_MAX_OPERATION_ATTEMPTS = 30; + +export async function* runPendingOperation( + step: StepRunner, + key: string, + operation: (context?: ResourceOperationContext) => T | Promise, + maxAttempts = DEFAULT_MAX_OPERATION_ATTEMPTS, +): AsyncGenerator { + if (!Number.isInteger(maxAttempts) || maxAttempts < 1) { + throw new RangeError("maxOperationAttempts must be a positive integer"); + } + + let context: ResourceOperationContext | undefined; + let pending: ResourceOperationPendingError | undefined; + + for (let attempt = 0; attempt < maxAttempts; attempt += 1) { + try { + return yield* step.run(`${key}:attempt:${attempt}`, () => + operation(context), + ); + } catch (error) { + if (!ResourceOperationPendingError.is(error)) throw error; + + pending = error; + context = error.callbackContext; + + if (attempt + 1 < maxAttempts) { + yield* step.delay(`${key}:retry-delay:${attempt}`, error.retryAfterMs); + } + } + } + + throw new Error( + `${pending?.message ?? "Resource operation remained pending"} after ${maxAttempts} attempts`, + { cause: pending }, + ); +} diff --git a/packages/reconciler/src/operations/operation.read.ts b/packages/reconciler/src/operations/operation.read.ts index 79db1de..e74196a 100644 --- a/packages/reconciler/src/operations/operation.read.ts +++ b/packages/reconciler/src/operations/operation.read.ts @@ -1,31 +1,11 @@ import { createWorkflow } from "yieldstar"; import { - DEFAULT_READ_POLL_OPTIONS, type ReadResourceParams, type StepRunner, emitLifecycleEvent, getErrorDetails, } from "./operation.types"; - -type ReadRetryCondition = { - key: string; - reason: string; - value?: unknown; -}; - -function needsReadRetry( - readResult: Record, - retryConditions: ReadonlyArray, -) { - return retryConditions.find((condition) => { - const resultValue = readResult[condition.key]; - if (condition.value !== undefined) { - return resultValue !== condition.value; - } - - return !resultValue; - }); -} +import { runPendingOperation } from "./operation.pending"; export async function* readResourceOperation( step: StepRunner, @@ -58,29 +38,16 @@ export async function* readResourceOperation( return merged as Record; } - let remoteOutput: Record = {}; - const retryConditions = (params.resource.retryReadOnCondition ?? []).filter( - Boolean, - ) as ReadRetryCondition[]; - - if (retryConditions.length > 0) { - yield* step.poll( - "read:poll-until-settled", - params.readPollOptions ?? DEFAULT_READ_POLL_OPTIONS, - async () => { - remoteOutput = await params.resource.read!(params.resource.key); - return !needsReadRetry(remoteOutput, retryConditions); - }, - ); - } else { - remoteOutput = yield* step.run("read:remote", () => - params.resource.read!(params.resource.key), - ); - } + const remote = yield* runPendingOperation( + step, + "read:remote", + (context) => params.resource.read!(params.resource.key, context), + params.maxOperationAttempts, + ); const mergedOutput = { ...resourceParams, - ...remoteOutput, + ...remote, }; await emitLifecycleEvent(params, "read", "success"); diff --git a/packages/reconciler/src/operations/operation.types.ts b/packages/reconciler/src/operations/operation.types.ts index 3def119..6cbcd1c 100644 --- a/packages/reconciler/src/operations/operation.types.ts +++ b/packages/reconciler/src/operations/operation.types.ts @@ -1,8 +1,4 @@ -import type { - BaseResource, - ErrorMatcher, - ResourceType, -} from "@notation/resource"; +import type { BaseResource, ResourceType } from "@notation/resource"; import type { State } from "@notation/state"; export type OperationName = "create" | "read" | "update" | "delete"; @@ -26,26 +22,12 @@ export type OperationEventEmitter = ( event: OperationLifecycleEvent, ) => void | Promise; -export type PollOptions = { - maxAttempts: number; - retryInterval: number; -}; - export type StepRunner = { run(fn: () => T | Promise): AsyncGenerator; run( key: string, fn: () => T | Promise, ): AsyncGenerator; - poll( - opts: PollOptions, - predicate: () => boolean | Promise, - ): AsyncGenerator; - poll( - key: string, - opts: PollOptions, - predicate: () => boolean | Promise, - ): AsyncGenerator; delay(ms: number): AsyncGenerator; delay(key: string, ms: number): AsyncGenerator; }; @@ -55,8 +37,7 @@ export type ResourceOperationBaseParams = { state: Pick; dryRun?: boolean; emit?: OperationEventEmitter; - retryOptions?: PollOptions; - readPollOptions?: PollOptions; + maxOperationAttempts?: number; }; export type CreateResourceParams = ResourceOperationBaseParams & { @@ -74,38 +55,6 @@ export type DeleteResourceParams = ResourceOperationBaseParams & { expectedRev: number; }; -export const DEFAULT_RETRY_OPTIONS: PollOptions = { - maxAttempts: 10, - retryInterval: 1_000, -}; - -export const DEFAULT_READ_POLL_OPTIONS: PollOptions = { - maxAttempts: 30, - retryInterval: 1_000, -}; - -export function matchError( - err: unknown, - matchers: ErrorMatcher[] | undefined, -): ErrorMatcher | undefined { - if (!matchers || matchers.length === 0) return undefined; - - const name = - typeof err === "object" && err && "name" in err - ? String((err as { name?: unknown }).name) - : undefined; - const message = - typeof err === "object" && err && "message" in err - ? String((err as { message?: unknown }).message) - : undefined; - - return matchers.find((matcher) => { - if (matcher.name !== name) return false; - if (matcher.message && matcher.message !== message) return false; - return true; - }); -} - export function getErrorDetails(err: unknown): { errorName: string; errorMessage: string; diff --git a/packages/reconciler/src/operations/operation.update.ts b/packages/reconciler/src/operations/operation.update.ts index 55dcba1..fc62a8c 100644 --- a/packages/reconciler/src/operations/operation.update.ts +++ b/packages/reconciler/src/operations/operation.update.ts @@ -1,12 +1,11 @@ -import { RetryableError, createWorkflow } from "yieldstar"; +import { createWorkflow } from "yieldstar"; import { - DEFAULT_RETRY_OPTIONS, type StepRunner, type UpdateResourceParams, emitLifecycleEvent, getErrorDetails, - matchError, } from "./operation.types"; +import { runPendingOperation } from "./operation.pending"; import { readResourceOperation } from "./operation.read"; export async function* updateResourceOperation( @@ -33,24 +32,19 @@ export async function* updateResourceOperation( params.resource.getParams(), ); - yield* step.run("update:remote", async () => { - try { - await params.resource.update!( + yield* runPendingOperation( + step, + "update:remote", + (context) => + params.resource.update!( params.resource.key, params.patch, resourceParams, params.resource.toState(params.resource.output), - ); - } catch (err) { - const matcher = matchError(err, params.resource.retryLaterOnError); - if (matcher) { - throw new RetryableError(matcher.reason, { - ...(params.retryOptions ?? DEFAULT_RETRY_OPTIONS), - }); - } - throw err; - } - }); + context, + ), + params.maxOperationAttempts, + ); params.resource.setOutput({ ...params.resource.key, @@ -61,7 +55,7 @@ export async function* updateResourceOperation( resource: params.resource, state: params.state, emit: params.emit, - readPollOptions: params.readPollOptions, + maxOperationAttempts: params.maxOperationAttempts, }); params.resource.setOutput({ diff --git a/packages/reconciler/src/plan.ts b/packages/reconciler/src/plan.ts index 87fb621..d2f8451 100644 --- a/packages/reconciler/src/plan.ts +++ b/packages/reconciler/src/plan.ts @@ -35,8 +35,7 @@ export type Plan = { }; export type DriftRead = - | { status: "found"; output: Record } - | { status: "not-found" }; + { kind: "present"; output: Record } | { kind: "absent" }; export type ResourceAction = | { decision: "create" } @@ -64,7 +63,7 @@ export function decideAction(opts: { >; if (driftRead) { - if (driftRead.status === "not-found") { + if (driftRead.kind === "absent") { return { decision: stateNode ? "drift-recreate" : "create" }; } @@ -115,22 +114,19 @@ export function decideAction(opts: { export async function resolvePlanParams( resource: BaseResource, ): Promise> { - let resolved: Record | undefined; - try { - resolved = (await resource.getParams()) as Record; - } catch { - resolved = undefined; - } - - const params: Record = {}; - - if (resolved) { + const hasUnresolvedDependency = Object.values(resource.dependencies).some( + (dependency) => dependency && dependency.output == null, + ); + if (!hasUnresolvedDependency) { + const resolved = (await resource.getParams()) as Record; + const params: Record = {}; for (const [key, value] of Object.entries(resolved)) { params[key] = value === undefined ? UNKNOWN_AFTER_APPLY : value; } return params; } + const params: Record = {}; const config = resource.config as Record; for (const [key, item] of Object.entries(resource.schema)) { if (item.propertyType === "computed") continue; diff --git a/packages/reconciler/src/reconciler.ts b/packages/reconciler/src/reconciler.ts index 2db4628..fab6c0b 100644 --- a/packages/reconciler/src/reconciler.ts +++ b/packages/reconciler/src/reconciler.ts @@ -1,6 +1,6 @@ +import { ResourceNotFoundError } from "@notation/resource"; import type { BaseResource, ResourceType } from "@notation/resource"; import { RevConflict, type State, type StateNode } from "@notation/state"; -import { RetryableError } from "yieldstar"; import { setTimeout as sleep } from "node:timers/promises"; import { buildResourceDepthLevels } from "./dependency-graph"; import { @@ -15,10 +15,8 @@ import { import { createResourceOperation, deleteResourceOperation, - matchError, readResourceOperation, type OperationLifecycleEvent, - type PollOptions, type StepRunner, updateResourceOperation, } from "./operations"; @@ -67,8 +65,7 @@ export type ReconcilerOptions = { dryRun?: boolean; driftDetection?: boolean; emit?: ReconcilerEventEmitter; - retryOptions?: PollOptions; - readPollOptions?: PollOptions; + maxOperationAttempts?: number; mutationLeaseTtl?: number; }; @@ -95,8 +92,7 @@ export class Reconciler { readonly #defaultDryRun: boolean; readonly #defaultDriftDetection: boolean; readonly #emit?: ReconcilerEventEmitter; - readonly #retryOptions?: PollOptions; - readonly #readPollOptions?: PollOptions; + readonly #maxOperationAttempts?: number; readonly #mutationLeaseTtl: number; readonly #stepRunner: StepRunner; @@ -106,8 +102,7 @@ export class Reconciler { this.#defaultDryRun = opts.dryRun ?? false; this.#defaultDriftDetection = opts.driftDetection ?? true; this.#emit = opts.emit; - this.#retryOptions = opts.retryOptions; - this.#readPollOptions = opts.readPollOptions; + this.#maxOperationAttempts = opts.maxOperationAttempts; this.#mutationLeaseTtl = opts.mutationLeaseTtl ?? 30_000; this.#stepRunner = createStepRunner(); } @@ -316,8 +311,7 @@ export class Reconciler { state: this.#state, dryRun, emit: this.#emit, - retryOptions: this.#retryOptions, - readPollOptions: this.#readPollOptions, + maxOperationAttempts: this.#maxOperationAttempts, expectedRev: stateNode?.rev ?? 0, }), ); @@ -332,8 +326,7 @@ export class Reconciler { patch: action.patch, dryRun, emit: this.#emit, - retryOptions: this.#retryOptions, - readPollOptions: this.#readPollOptions, + maxOperationAttempts: this.#maxOperationAttempts, expectedRev: stateNode!.rev, }), ); @@ -361,7 +354,7 @@ export class Reconciler { params, driftRead: remote, }); - if (remote.status === "found") resource.setOutput(remote.output); + if (remote.kind === "present") resource.setOutput(remote.output); await this.#emit?.({ level: "info", @@ -380,8 +373,7 @@ export class Reconciler { state: this.#state, dryRun, emit: this.#emit, - retryOptions: this.#retryOptions, - readPollOptions: this.#readPollOptions, + maxOperationAttempts: this.#maxOperationAttempts, expectedRev: stateNode?.rev ?? 0, }), ); @@ -395,8 +387,7 @@ export class Reconciler { patch: action.patch, dryRun, emit: this.#emit, - retryOptions: this.#retryOptions, - readPollOptions: this.#readPollOptions, + maxOperationAttempts: this.#maxOperationAttempts, expectedRev: stateNode?.rev ?? 0, }), ); @@ -452,14 +443,13 @@ export class Reconciler { resource, state: this.#state, emit: this.#emit, - readPollOptions: this.#readPollOptions, + maxOperationAttempts: this.#maxOperationAttempts, }), ); - return { status: "found", output }; - } catch (err) { - const matcher = matchError(err, resource.notFoundOnError); - if (!matcher) throw err; - return { status: "not-found" }; + return { kind: "present", output }; + } catch (error) { + if (ResourceNotFoundError.is(error)) return { kind: "absent" }; + throw error; } } @@ -537,7 +527,7 @@ export class Reconciler { if (!resource.read) throw conflict; const remote = await this.#readForDrift(resource); - if (remote.status === "not-found") { + if (remote.kind !== "present") { if (!dryRun) await this.#state.delete(resource.id, stateNode.rev); return; } @@ -550,7 +540,7 @@ export class Reconciler { state: this.#state, dryRun, emit: this.#emit, - retryOptions: this.#retryOptions, + maxOperationAttempts: this.#maxOperationAttempts, expectedRev: stateNode.rev, }), ); @@ -595,37 +585,7 @@ export function createStepRunner(): StepRunner { throw new Error("Missing run function"); } - while (true) { - try { - return await fn(); - } catch (err) { - if (!(err instanceof RetryableError)) { - throw err; - } - } - } - }, - async *poll( - arg1: string | PollOptions, - arg2: PollOptions | (() => boolean | Promise), - arg3?: () => boolean | Promise, - ): AsyncGenerator { - const opts = (typeof arg1 === "string" ? arg2 : arg1) as PollOptions; - const predicate = (typeof arg1 === "string" ? arg3 : arg2) as - (() => boolean | Promise) | undefined; - - if (!predicate) { - throw new Error("Missing poll predicate"); - } - - for (let attempt = 0; attempt < opts.maxAttempts; attempt += 1) { - if (await predicate()) return; - } - - throw new RetryableError("Polling reached max retries", { - maxAttempts: opts.maxAttempts, - retryInterval: opts.retryInterval, - }); + return await fn(); }, async *delay( arg1: string | number, diff --git a/packages/reconciler/test/operation.workflows.test.ts b/packages/reconciler/test/operation.workflows.test.ts index 8709086..2afc6df 100644 --- a/packages/reconciler/test/operation.workflows.test.ts +++ b/packages/reconciler/test/operation.workflows.test.ts @@ -1,12 +1,14 @@ import { describe, expect, it, vi } from "vitest"; -import { RetryableError } from "yieldstar"; -import { resource } from "@notation/resource"; +import { + resource, + ResourceNotFoundError, + ResourceOperationPendingError, +} from "@notation/resource"; import { createResourceOperation, deleteResourceOperation, readResourceOperation, type OperationLifecycleEvent, - type PollOptions, type StepRunner, } from "../src/operations"; @@ -20,50 +22,19 @@ function createStepRunnerDouble(): StepRunner { throw new Error("Missing run function"); } - while (true) { - try { - return await fn(); - } catch (err) { - if (!(err instanceof RetryableError)) { - throw err; - } - } - } - }); - - const poll = vi.fn(async function* ( - arg1: string | PollOptions, - arg2: PollOptions | (() => boolean | Promise), - arg3?: () => boolean | Promise, - ): AsyncGenerator { - const opts = (typeof arg1 === "string" ? arg2 : arg1) as PollOptions; - const predicate = (typeof arg1 === "string" ? arg3 : arg2) as - | (() => boolean | Promise) - | undefined; - - if (!predicate) { - throw new Error("Missing poll predicate"); - } - - for (let attempt = 0; attempt < opts.maxAttempts; attempt++) { - if (await predicate()) { - return; - } - } - - throw new RetryableError("Polling reached max retries", { - maxAttempts: opts.maxAttempts, - retryInterval: opts.retryInterval, - }); + return await fn(); }); - const delay = vi.fn(async function* (): AsyncGenerator { + const delay = vi.fn(async function* (): AsyncGenerator< + unknown, + void, + unknown + > { return; }); return { run, - poll, delay, }; } @@ -87,13 +58,17 @@ describe("operation workflows", () => { }; let createAttempts = 0; - const createMock = vi.fn(async () => { + const createMock = vi.fn(async (_params, context) => { createAttempts += 1; if (createAttempts === 1) { - const err = new Error("eventual consistency"); - err.name = "RetryCreate"; - throw err; + expect(context).toBeUndefined(); + throw Object.assign(new Error("retry create"), { + _tag: "ResourceOperationPendingError", + retryAfterMs: 25, + callbackContext: { operationId: "create-123" }, + }); } + expect(context).toEqual({ operationId: "create-123" }); return { remoteId: "abc" }; }); @@ -103,7 +78,6 @@ describe("operation workflows", () => { create: createMock, read: async () => ({ remoteId: "abc", status: "ready" }), delete: async () => undefined, - retryLaterOnError: [{ name: "RetryCreate", reason: "retry create" }], }); const testResource = new TestResource({ id: "test-create" }); @@ -121,14 +95,20 @@ describe("operation workflows", () => { expect(createAttempts).toBe(2); expect(state.update).toHaveBeenCalledOnce(); - expect(createMock).toHaveBeenCalledWith(await testResource.getParams()); + expect(createMock).toHaveBeenNthCalledWith( + 1, + await testResource.getParams(), + undefined, + ); + expect(createMock).toHaveBeenNthCalledWith( + 2, + await testResource.getParams(), + { operationId: "create-123" }, + ); expect(testResource.output).toEqual({ remoteId: "abc", status: "ready" }); - expect(events.map((event) => `${event.operation}:${event.status}`)).toEqual([ - "create:start", - "read:start", - "read:success", - "create:success", - ]); + expect(events.map((event) => `${event.operation}:${event.status}`)).toEqual( + ["create:start", "read:start", "read:success", "create:success"], + ); expect(events[0]).toMatchObject({ resourceId: "test-create", resourceType: TestResource.type, @@ -136,7 +116,7 @@ describe("operation workflows", () => { }); }); - it("read uses durable polling semantics for retryReadOnCondition", async () => { + it("read follows pending retry instructions", async () => { const step = createStepRunnerDouble(); const state = { get: vi.fn(async () => undefined), @@ -149,21 +129,18 @@ describe("operation workflows", () => { .defineSchema({}) .defineOperations({ create: async () => ({}), - read: async () => { + read: async (_key, context) => { readAttempts += 1; if (readAttempts < 3) { - return { status: "pending" }; + throw new ResourceOperationPendingError("resource is not ready", { + retryAfterMs: readAttempts * 10, + callbackContext: { readAttempts }, + }); } - return { status: "ready" }; + expect(context).toEqual({ readAttempts: 2 }); + return { status: "ready" } as const; }, delete: async () => undefined, - retryReadOnCondition: [ - { - key: "status", - value: "ready", - reason: "resource is not ready", - }, - ], }); const testResource = new TestResource({ id: "test-read" }); @@ -176,11 +153,82 @@ describe("operation workflows", () => { ); expect(readAttempts).toBe(3); - expect((step.poll as any).mock.calls.length).toBe(1); expect(result).toEqual({ status: "ready" }); + expect(step.delay).toHaveBeenNthCalledWith( + 1, + "read:remote:retry-delay:0", + 10, + ); + expect(step.delay).toHaveBeenNthCalledWith( + 2, + "read:remote:retry-delay:1", + 20, + ); + }); + + it("fails when an operation remains pending past the safety limit", async () => { + const step = createStepRunnerDouble(); + const state = { + get: vi.fn(async () => undefined), + update: vi.fn(async () => undefined), + delete: vi.fn(async () => undefined), + }; + const read = vi.fn(async () => { + throw new ResourceOperationPendingError("still pending", { + retryAfterMs: 10, + }); + }); + const TestResource = resource({ type: "test/service/pending-limit" }) + .defineSchema({}) + .defineOperations({ + create: async () => ({}), + read, + delete: async () => undefined, + }); + + await expect( + runOperation( + readResourceOperation(step, { + resource: new TestResource({ id: "pending-limit" }), + state, + maxOperationAttempts: 2, + }), + ), + ).rejects.toThrowError("still pending after 2 attempts"); + expect(read).toHaveBeenCalledTimes(2); + expect(step.delay).toHaveBeenCalledOnce(); }); - it("delete treats only resource.notFoundOnError matchers as skip", async () => { + it("does not infer that not-found after creation is retryable", async () => { + const step = createStepRunnerDouble(); + const state = { + get: vi.fn(async () => undefined), + update: vi.fn(async () => undefined), + delete: vi.fn(async () => undefined), + }; + const TestResource = resource({ type: "test/service/eventually-visible" }) + .defineSchema({}) + .defineOperations({ + create: async () => ({}), + read: async () => { + throw new ResourceNotFoundError("resource is absent"); + }, + delete: async () => undefined, + }); + + await expect( + runOperation( + createResourceOperation(step, { + resource: new TestResource({ id: "eventually-visible" }), + state, + expectedRev: 0, + }), + ), + ).rejects.toThrowError("resource is absent"); + expect(state.update).not.toHaveBeenCalled(); + }); + + it("delete treats an already-absent remote as success through its idempotent resource contract", async () => { const step = createStepRunnerDouble(); const events: OperationLifecycleEvent[] = []; const state = { @@ -193,17 +241,7 @@ describe("operation workflows", () => { .defineSchema({}) .defineOperations({ create: async () => ({}), - delete: async () => { - const err = new Error("gone"); - err.name = "RemoteMissing"; - throw err; - }, - notFoundOnError: [ - { - name: "RemoteMissing", - reason: "already deleted remotely", - }, - ], + delete: async () => undefined, }); const testResource = new TestResource({ id: "test-delete" }); @@ -220,14 +258,10 @@ describe("operation workflows", () => { ); expect(state.delete).toHaveBeenCalledWith("test-delete", 1); - expect(events.map((event) => event.status)).toEqual([ - "start", - "skip", - "success", - ]); + expect(events.map((event) => event.status)).toEqual(["start", "success"]); }); - it("delete rethrows when error does not match notFoundOnError", async () => { + it("delete rethrows an unclassified resource error", async () => { const step = createStepRunnerDouble(); const state = { get: vi.fn(async () => undefined), @@ -244,12 +278,6 @@ describe("operation workflows", () => { err.name = "DifferentError"; throw err; }, - notFoundOnError: [ - { - name: "RemoteMissing", - reason: "already deleted remotely", - }, - ], }); const testResource = new TestResource({ id: "test-delete-miss" }); @@ -262,7 +290,10 @@ describe("operation workflows", () => { expectedRev: 1, }), ), - ).rejects.toMatchObject({ name: "DifferentError", message: "still exists" }); + ).rejects.toMatchObject({ + name: "DifferentError", + message: "still exists", + }); expect(state.delete).not.toHaveBeenCalled(); }); diff --git a/packages/reconciler/test/reconciler.deploy.test.ts b/packages/reconciler/test/reconciler.deploy.test.ts index 7220a65..eaa5ce5 100644 --- a/packages/reconciler/test/reconciler.deploy.test.ts +++ b/packages/reconciler/test/reconciler.deploy.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from "vitest"; -import { resource, type ErrorMatcher } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; import { LeaseConflict, MemoryStateBackend, @@ -70,7 +70,6 @@ function createTestResourceClass(opts: { key: Record, state: Record, ) => Promise; - notFoundOnError?: ErrorMatcher[]; }) { return resource({ type: opts.type }) .defineSchema({ @@ -85,11 +84,11 @@ function createTestResourceClass(opts: { read: opts.read, update: opts.update, delete: opts.delete ?? (async () => undefined), - notFoundOnError: opts.notFoundOnError, }); } const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); +const found = (output: Record) => output; describe("reconciler deploy", () => { it("chooses create vs update from desired params vs state", async () => { @@ -99,12 +98,12 @@ describe("reconciler deploy", () => { const CreateResource = createTestResourceClass({ type: "test/service/create-choice", create: createSpy, - read: async () => ({ name: "new" }), + read: async () => found({ name: "new" }), }); const UpdateResource = createTestResourceClass({ type: "test/service/update-choice", update: updateSpy, - read: async () => ({ name: "new" }), + read: async () => found({ name: "new" }), }); const state = createMemoryState({ @@ -149,7 +148,7 @@ describe("reconciler deploy", () => { const CreateResource = createTestResourceClass({ type: "test/service/first-create", create: async () => ({ name: "new" }), - read: async () => ({ name: "new" }), + read: async () => found({ name: "new" }), }); const state = createMemoryState(); const reconciler = new Reconciler({ state, driftDetection: false }); @@ -178,7 +177,7 @@ describe("reconciler deploy", () => { const CreateResource = createTestResourceClass({ type: "test/service/concurrent-create", create: createSpy, - read: async () => ({ name: "new" }), + read: async () => found({ name: "new" }), }); const state = new MemoryStateBackend(); const first = new Reconciler({ state, driftDetection: false }); @@ -202,7 +201,7 @@ describe("reconciler deploy", () => { it("reads remote state after an update conflict instead of repeating the update", async () => { let remoteName = "old"; - const readSpy = vi.fn(async () => ({ name: remoteName })); + const readSpy = vi.fn(async () => found({ name: remoteName })); const updateSpy = vi.fn(async (_key, _patch, params) => { remoteName = params.name as string; }); @@ -268,7 +267,7 @@ describe("reconciler deploy", () => { remoteName = params.name as string; return { name: remoteName }; }); - const readSpy = vi.fn(async () => ({ name: remoteName! })); + const readSpy = vi.fn(async () => found({ name: remoteName! })); const CreateResource = createTestResourceClass({ type: "test/service/create-conflict", create: createSpy, @@ -360,7 +359,7 @@ describe("reconciler deploy", () => { marks.aEnd = Date.now(); return { name: "a" }; }, - read: async () => ({ name: "a" }), + read: async () => found({ name: "a" }), }); const CResource = createTestResourceClass({ type: "test/service/c", @@ -370,7 +369,7 @@ describe("reconciler deploy", () => { marks.cEnd = Date.now(); return { name: "c" }; }, - read: async () => ({ name: "c" }), + read: async () => found({ name: "c" }), }); const BResource = createTestResourceClass({ type: "test/service/b", @@ -378,7 +377,7 @@ describe("reconciler deploy", () => { marks.bStart = Date.now(); return { name: "b" }; }, - read: async () => ({ name: "b" }), + read: async () => found({ name: "b" }), }); const state = createMemoryState(); @@ -402,7 +401,7 @@ describe("reconciler deploy", () => { const events: Array> = []; const TestResource = createTestResourceClass({ type: "test/service/drift", - read: async () => ({ name: "drifted" }), + read: async () => found({ name: "drifted" }), update: updateSpy, }); @@ -484,7 +483,7 @@ describe("reconciler deploy", () => { const CreateResource = createTestResourceClass({ type: "test/service/dry-run-create", create: createSpy, - read: async () => ({ name: "new" }), + read: async () => found({ name: "new" }), }); const OrphanResource = createTestResourceClass({ type: "test/service/dry-run-orphan", @@ -542,17 +541,14 @@ describe("reconciler destroy + refresh", () => { }); const readSpy = vi.fn(async () => { if (!remoteExists) { - const error = new Error("gone"); - error.name = "RemoteMissing"; - throw error; + throw new ResourceNotFoundError("resource is absent"); } - return { name: "doomed" }; + return found({ name: "doomed" }); }); const DestroyResource = createTestResourceClass({ type: "test/service/destroy-retry", read: readSpy, delete: deleteSpy, - notFoundOnError: [{ name: "RemoteMissing", reason: "deleted" }], }); const state = createMemoryState({ doomed: { diff --git a/packages/reconciler/test/reconciler.plan.test.ts b/packages/reconciler/test/reconciler.plan.test.ts index 88b8c91..d98e55a 100644 --- a/packages/reconciler/test/reconciler.plan.test.ts +++ b/packages/reconciler/test/reconciler.plan.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it, vi } from "vitest"; -import { resource, type BaseResource } from "@notation/resource"; +import { + ResourceNotFoundError, + ResourceOperationPendingError, + resource, + type BaseResource, +} from "@notation/resource"; import type { StateNode } from "@notation/state"; import { Reconciler, UNKNOWN_AFTER_APPLY } from "../src"; @@ -48,7 +53,6 @@ function createTestResourceClass(opts: { key: Record, state: Record, ) => Promise; - notFoundOnError?: { name: string; reason: string }[]; }) { return resource({ type: opts.type }) .defineSchema({ @@ -68,7 +72,6 @@ function createTestResourceClass(opts: { read: opts.read, update: opts.update, delete: opts.delete ?? (async () => undefined), - notFoundOnError: opts.notFoundOnError, }); } @@ -204,13 +207,8 @@ describe("reconciler plan", () => { const TestResource = createTestResourceClass({ type: "test/service/plan-drift-recreate", read: async () => { - const err = new Error("gone"); - err.name = "NotFoundException"; - throw err; + throw new ResourceNotFoundError("resource is absent"); }, - notFoundOnError: [ - { name: "NotFoundException", reason: "deleted remotely" }, - ], }); const state = createMemoryState({ @@ -232,6 +230,40 @@ describe("reconciler plan", () => { }); }); + it("waits for a pending read before planning", async () => { + let attempts = 0; + const TestResource = createTestResourceClass({ + type: "test/service/plan-not-ready", + read: async () => { + attempts += 1; + if (attempts === 1) { + throw new ResourceOperationPendingError( + "Waiting for Lambda to become active", + { retryAfterMs: 0 }, + ); + } + return { name: "desired" }; + }, + }); + + const state = createMemoryState({ + resource: createStateNode("resource", "test/service/plan-not-ready", { + name: "desired", + }), + }); + const reconciler = new Reconciler({ state, driftDetection: true }); + + const plan = await reconciler.plan([ + new TestResource({ id: "resource", config: { name: "desired" } }), + ]); + + expect(plan.nodes[0]).toMatchObject({ + id: "resource", + decision: "noop", + }); + expect(attempts).toBe(2); + }); + it("skips remote reads when drift detection is off", async () => { const readSpy = vi.fn(async () => ({ name: "drifted" })); const TestResource = createTestResourceClass({ @@ -332,6 +364,35 @@ describe("reconciler plan", () => { }); }); + it("does not disguise parameter derivation failures as unknown values", async () => { + const TestResource = resource({ + type: "test/service/plan-derive-failure", + }) + .defineSchema({ + name: { + presence: "required", + propertyType: "param", + valueType: "string" as any, + }, + }) + .defineOperations({ + create: async () => ({}), + delete: async () => undefined, + deriveParams: () => { + throw new Error("invalid derived configuration"); + }, + }); + + const reconciler = new Reconciler({ + state: createMemoryState(), + driftDetection: false, + }); + + await expect( + reconciler.plan([new TestResource({ id: "broken" })]), + ).rejects.toThrow("invalid derived configuration"); + }); + it("produces a JSON-round-trippable plan", async () => { const CreateResource = createTestResourceClass({ type: "test/service/plan-json-create", diff --git a/packages/resource/src/index.ts b/packages/resource/src/index.ts index 46a899b..f531b96 100644 --- a/packages/resource/src/index.ts +++ b/packages/resource/src/index.ts @@ -1,6 +1,7 @@ export * from "./types"; export * from "./resource.schema"; export * from "./resource"; +export * from "./resource-operation"; export * from "./resource-group"; export { collectResourceGraph, diff --git a/packages/resource/src/resource-operation.ts b/packages/resource/src/resource-operation.ts new file mode 100644 index 0000000..897dea4 --- /dev/null +++ b/packages/resource/src/resource-operation.ts @@ -0,0 +1,74 @@ +/** Plain data passed from one attempt of an operation to the next. */ +export type ResourceOperationContext = Readonly>; + +/** + * A `read` operation throws this error when no resource exists for its key. + */ +export class ResourceNotFoundError extends Error { + readonly _tag = "ResourceNotFoundError" as const; + + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "ResourceNotFoundError"; + } + + static is(error: unknown): error is ResourceNotFoundError { + return hasTag(error, "ResourceNotFoundError"); + } +} + +/** + * An operation throws this error when it has not finished. + * + * The reconciler waits for `retryAfterMs`, then calls the same operation again + * with `callbackContext`. + */ +export class ResourceOperationPendingError extends Error { + readonly _tag = "ResourceOperationPendingError" as const; + + /** Milliseconds to wait before the next attempt. */ + readonly retryAfterMs: number; + + /** Plain data passed to the next attempt of the same operation. */ + readonly callbackContext?: ResourceOperationContext; + + constructor( + message: string, + options: ErrorOptions & { + retryAfterMs: number; + callbackContext?: ResourceOperationContext; + }, + ) { + super(message, options); + this.name = "ResourceOperationPendingError"; + this.retryAfterMs = options.retryAfterMs; + this.callbackContext = options.callbackContext; + + if (!Number.isFinite(options.retryAfterMs) || options.retryAfterMs < 0) { + throw new RangeError("retryAfterMs must be a non-negative number"); + } + } + + static is(error: unknown): error is ResourceOperationPendingError { + return ( + hasTag(error, "ResourceOperationPendingError") && + "retryAfterMs" in error && + typeof error.retryAfterMs === "number" + ); + } +} + +export type ResourceOperationSignal = + ResourceNotFoundError | ResourceOperationPendingError; + +function hasTag( + error: unknown, + tag: T, +): error is Error & { _tag: T } { + return ( + typeof error === "object" && + error !== null && + "_tag" in error && + error._tag === tag + ); +} diff --git a/packages/resource/src/resource.ts b/packages/resource/src/resource.ts index b2ed61d..f191059 100644 --- a/packages/resource/src/resource.ts +++ b/packages/resource/src/resource.ts @@ -17,27 +17,12 @@ import type { Fallback, NoInfer, } from "./types"; +import type { ResourceOperationContext } from "./resource-operation"; export type { Schema, SchemaItem, DefineResourceApiSchema }; export type ResourceType = `${string}/${string}/${string}`; -export type ErrorMatcher = { - name: string; - message?: string; - reason: string; -}; - -export type ResultCondition = { - key: K; - reason: string; - value?: T[K]; -}; - -export type ResultConditions = { - [K in keyof T]?: ResultCondition; -}[keyof T][]; - export type ResourceOpts = OptionalIfAllPropertiesOptional<"config", C> & OptionalIfAllPropertiesOptional<"dependencies", D> & { id: string }; @@ -77,19 +62,27 @@ export interface BaseResource { groupType: string; readonly output: {}; readonly dependencies: Record; - readonly retryReadOnCondition?: ({ - key: any; - value?: any; - reason: string; - } | void)[]; - readonly failOnError?: (ErrorMatcher & { reason: string })[]; - readonly notFoundOnError?: ErrorMatcher[]; - readonly retryLaterOnError?: ErrorMatcher[]; readonly key: {}; - create: (params: any) => Promise<{} | void>; - read?: (key: any) => Promise>; - update?: (key: any, patch: any, params: any, state: any) => Promise; - delete: (key: any, state: any) => Promise; + create: ( + params: any, + context?: ResourceOperationContext, + ) => Promise<{} | void>; + read?: ( + key: any, + context?: ResourceOperationContext, + ) => Promise>; + update?: ( + key: any, + patch: any, + params: any, + state: any, + context?: ResourceOperationContext, + ) => Promise; + delete: ( + key: any, + state: any, + context?: ResourceOperationContext, + ) => Promise; getParams(): Promise<{}>; toState(output: {}): {}; toComparable(output: {}): {}; @@ -114,19 +107,26 @@ export abstract class Resource< dependencies = {} as NoInfer; abstract type: ResourceType; abstract schema: Schema; - abstract create: (params: T["params"]) => Promise; - abstract read?: (key: T["compoundKey"]) => Promise; + abstract create: ( + params: T["params"], + context?: ResourceOperationContext, + ) => Promise; + abstract read?: ( + key: T["compoundKey"], + context?: ResourceOperationContext, + ) => Promise; abstract update?: ( key: T["compoundKey"], patch: T["params"], params: T["params"], state: T["state"], + context?: ResourceOperationContext, + ) => Promise; + abstract delete: ( + key: T["compoundKey"], + state: T["state"], + context?: ResourceOperationContext, ) => Promise; - abstract delete: (key: T["compoundKey"], state: T["state"]) => Promise; - abstract retryReadOnCondition?: ResultConditions; - abstract failOnError?: (ErrorMatcher & { reason: string })[]; - abstract notFoundOnError?: ErrorMatcher[]; - abstract retryLaterOnError?: ErrorMatcher[]; abstract deriveParams(opts: { id: string; config: C; @@ -198,19 +198,26 @@ export type ResourceOperationsOptions< T extends ResourceTypes, IntrinsicParams extends Partial, > = { - create: (params: T["params"]) => Promise; - read?: (key: T["compoundKey"]) => Promise; + create: ( + params: T["params"], + context?: ResourceOperationContext, + ) => Promise; + read?: ( + key: T["compoundKey"], + context?: ResourceOperationContext, + ) => Promise; update?: ( key: T["compoundKey"], patch: T["params"], params: T["params"], state: T["state"], + context?: ResourceOperationContext, + ) => Promise; + delete: ( + key: T["compoundKey"], + state: T["state"], + context?: ResourceOperationContext, ) => Promise; - delete: (key: T["compoundKey"], state: T["state"]) => Promise; - retryReadOnCondition?: ResultConditions; - failOnError?: (ErrorMatcher & { reason: string })[]; - notFoundOnError?: ErrorMatcher[]; - retryLaterOnError?: ErrorMatcher[]; deriveParams?: (opts: { config: Partial; }) => IntrinsicParams | Promise; @@ -315,10 +322,6 @@ export function defineResource( read = opts.read ? opts.read : undefined; update = opts.update ? opts.update : undefined; delete = opts.delete; - retryReadOnCondition = opts.retryReadOnCondition; - failOnError = opts.failOnError; - notFoundOnError = opts.notFoundOnError; - retryLaterOnError = opts.retryLaterOnError; async deriveParams() { if (!opts.deriveParams) return {}; diff --git a/packages/resource/test/resource.doubles.ts b/packages/resource/test/resource.doubles.ts index 1293849..d52dd7f 100644 --- a/packages/resource/test/resource.doubles.ts +++ b/packages/resource/test/resource.doubles.ts @@ -65,7 +65,11 @@ export const testOperations = { async delete() {}, async update() {}, async read() { - return { primaryKey: "", optionalSecondaryKey: "", requiredParam: "" }; + return { + primaryKey: "", + optionalSecondaryKey: "", + requiredParam: "", + } as const; }, deriveParams() { return { intrinsicParam: true }; diff --git a/packages/resource/test/resource.test.ts b/packages/resource/test/resource.test.ts index ca5722a..b3e3b7e 100644 --- a/packages/resource/test/resource.test.ts +++ b/packages/resource/test/resource.test.ts @@ -1,5 +1,9 @@ import { expect, it, test, vi } from "vitest"; -import { resource } from "src"; +import { + ResourceNotFoundError, + ResourceOperationPendingError, + resource, +} from "src"; import { TestResource, testResourceConfig, @@ -138,3 +142,52 @@ describe("resource dependencies", () => { }); }); }); + +describe("resource operation signals", () => { + it("recognises a not-found signal structurally", () => { + expect(ResourceNotFoundError.is(new ResourceNotFoundError("missing"))).toBe( + true, + ); + expect( + ResourceNotFoundError.is( + Object.assign(new Error("missing"), { + _tag: "ResourceNotFoundError", + }), + ), + ).toBe(true); + }); + + it("recognises a pending signal structurally", () => { + const fromElsewhere = Object.assign(new Error("waiting"), { + _tag: "ResourceOperationPendingError", + retryAfterMs: 2_000, + callbackContext: { operationId: "abc" }, + }); + expect(ResourceOperationPendingError.is(fromElsewhere)).toBe(true); + }); + + it("rejects unrelated errors", () => { + expect(ResourceNotFoundError.is(new Error("boom"))).toBe(false); + expect(ResourceOperationPendingError.is(new Error("boom"))).toBe(false); + expect( + ResourceOperationPendingError.is({ + _tag: "ResourceOperationPendingError", + }), + ).toBe(false); + }); + + it("carries retry instructions and callback context", () => { + const pending = new ResourceOperationPendingError("waiting", { + retryAfterMs: 1_500, + callbackContext: { operationId: "abc" }, + }); + expect(pending.retryAfterMs).toBe(1_500); + expect(pending.callbackContext).toEqual({ operationId: "abc" }); + }); + + it("rejects an invalid retry delay", () => { + expect( + () => new ResourceOperationPendingError("waiting", { retryAfterMs: -1 }), + ).toThrowError("retryAfterMs must be a non-negative number"); + }); +}); diff --git a/packages/state/package.json b/packages/state/package.json index 2c81b57..8a49f6f 100644 --- a/packages/state/package.json +++ b/packages/state/package.json @@ -11,6 +11,9 @@ "build": "tsup --clean", "dev": "tsup --watch" }, + "dependencies": { + "@notation/utils": "workspace:*" + }, "devDependencies": { "@types/node": "^22.13.4" } diff --git a/packages/state/src/state.ts b/packages/state/src/state.ts index 003faa1..d0c10c1 100644 --- a/packages/state/src/state.ts +++ b/packages/state/src/state.ts @@ -9,6 +9,7 @@ import { } from "node:fs/promises"; import path from "node:path"; import { setTimeout as sleep } from "node:timers/promises"; +import { isErrorWithCode } from "@notation/utils"; import { LeaseConflict, RevConflict } from "./conflicts"; export type StateNode = { @@ -222,7 +223,7 @@ export class FileStateBackend implements StateBackend { }); break; } catch (error) { - if (!isFileExistsError(error)) throw error; + if (!isErrorWithCode(error, "EEXIST")) throw error; const current = await readFileLease(leaseFilePath); if (!current || current.expiresAtMs <= Date.now()) { await unlink(leaseFilePath).catch(() => undefined); @@ -271,7 +272,7 @@ export class FileStateBackend implements StateBackend { const file = await readFile(this.stateFilePath, "utf8"); return JSON.parse(file) as Record; } catch (error) { - if (isFileMissingError(error)) { + if (isErrorWithCode(error, "ENOENT")) { return {}; } @@ -298,7 +299,7 @@ export class FileStateBackend implements StateBackend { ); break; } catch (error) { - if (!isFileExistsError(error)) throw error; + if (!isErrorWithCode(error, "EEXIST")) throw error; const lockStat = await stat(lockFilePath).catch(() => undefined); if (lockStat && Date.now() - lockStat.mtimeMs > FILE_LOCK_STALE_MS) { await unlink(lockFilePath).catch(() => undefined); @@ -367,29 +368,12 @@ async function readFileLease( try { return JSON.parse(await readFile(filePath, "utf8")) as FileLeaseRecord; } catch (error) { - if (isFileMissingError(error) || error instanceof SyntaxError) + if (isErrorWithCode(error, "ENOENT") || error instanceof SyntaxError) return undefined; throw error; } } -function isFileMissingError(error: unknown): boolean { - return isErrorWithCode(error, "ENOENT"); -} - -function isFileExistsError(error: unknown): boolean { - return isErrorWithCode(error, "EEXIST"); -} - -function isErrorWithCode(error: unknown, code: string): boolean { - return ( - typeof error === "object" && - error !== null && - "code" in error && - error.code === code - ); -} - function cloneAsPersistedState( state: Record, ): Record { diff --git a/packages/std.iac/package.json b/packages/std.iac/package.json index 63cfaa2..90d9789 100644 --- a/packages/std.iac/package.json +++ b/packages/std.iac/package.json @@ -15,6 +15,7 @@ "dependencies": { "@notation/core": "workspace:*", "@notation/resource": "workspace:*", + "@notation/utils": "workspace:*", "fflate": "0.8.2" }, "devDependencies": { diff --git a/packages/std.iac/src/resources/fs/file.ts b/packages/std.iac/src/resources/fs/file.ts index 52509a0..86c32cc 100644 --- a/packages/std.iac/src/resources/fs/file.ts +++ b/packages/std.iac/src/resources/fs/file.ts @@ -1,4 +1,5 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; +import { isErrorWithCode } from "@notation/utils"; import { getSourceSha256 } from "src/utils/hash"; import * as fs from "node:fs/promises"; @@ -36,8 +37,15 @@ export const File = fileSchema.defineOperations({ return { sourceSha256 }; }, read: async (config) => { - const file = await fs.readFile(config.filePath); - return { ...config, file }; + try { + const file = await fs.readFile(config.filePath); + return { ...config, file }; + } catch (error) { + if (isErrorWithCode(error, "ENOENT")) { + throw new ResourceNotFoundError("File was not found", { cause: error }); + } + throw error; + } }, create: async () => {}, update: async () => {}, diff --git a/packages/std.iac/src/resources/fs/zip.ts b/packages/std.iac/src/resources/fs/zip.ts index f8d5585..dae22d6 100644 --- a/packages/std.iac/src/resources/fs/zip.ts +++ b/packages/std.iac/src/resources/fs/zip.ts @@ -1,4 +1,5 @@ -import { resource } from "@notation/resource"; +import { resource, ResourceNotFoundError } from "@notation/resource"; +import { isErrorWithCode } from "@notation/utils"; import * as fs from "node:fs/promises"; import { zip } from "src/utils/zip"; import { getSourceSha256 } from "src/utils/hash"; @@ -54,11 +55,13 @@ export const Zip = zipSchema.defineOperations({ try { const file = await fs.readFile(params.filePath); return { ...params, file }; - } catch (error: any) { - if (error.code !== "ENOENT") throw error; - await zip.package(params.sourceFilePath, params.filePath); - const file = await fs.readFile(params.filePath); - return { ...params, file }; + } catch (error) { + if (isErrorWithCode(error, "ENOENT")) { + throw new ResourceNotFoundError("Zip archive was not found", { + cause: error, + }); + } + throw error; } }, create: async (params) => { @@ -69,7 +72,11 @@ export const Zip = zipSchema.defineOperations({ await zip.package(config.sourceFilePath, config.filePath); }, delete: async (config) => { - await fs.unlink(config.filePath); + try { + await fs.unlink(config.filePath); + } catch (error) { + if (!isErrorWithCode(error, "ENOENT")) throw error; + } }, }); diff --git a/packages/utils/package.json b/packages/utils/package.json new file mode 100644 index 0000000..ef2ec23 --- /dev/null +++ b/packages/utils/package.json @@ -0,0 +1,18 @@ +{ + "type": "module", + "name": "@notation/utils", + "version": "0.12.0", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "files": [ + "dist" + ], + "scripts": { + "build": "tsup --clean", + "dev": "tsup --watch", + "typecheck": "tsc --noEmit" + }, + "devDependencies": { + "@types/node": "^22.13.4" + } +} diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts new file mode 100644 index 0000000..fc8b93c --- /dev/null +++ b/packages/utils/src/index.ts @@ -0,0 +1,11 @@ +export function isErrorWithCode( + error: unknown, + code: string, +): error is { code: string } { + return ( + typeof error === "object" && + error !== null && + "code" in error && + error.code === code + ); +} diff --git a/packages/utils/test/index.test.ts b/packages/utils/test/index.test.ts new file mode 100644 index 0000000..99a969a --- /dev/null +++ b/packages/utils/test/index.test.ts @@ -0,0 +1,13 @@ +import { describe, expect, it } from "vitest"; +import { isErrorWithCode } from "../src"; + +describe("isErrorWithCode", () => { + it("recognises an object with the requested code", () => { + expect(isErrorWithCode({ code: "ENOENT" }, "ENOENT")).toBe(true); + }); + + it("rejects other codes and non-object values", () => { + expect(isErrorWithCode({ code: "EEXIST" }, "ENOENT")).toBe(false); + expect(isErrorWithCode("ENOENT", "ENOENT")).toBe(false); + }); +}); diff --git a/packages/utils/tsconfig.json b/packages/utils/tsconfig.json new file mode 100644 index 0000000..4423630 --- /dev/null +++ b/packages/utils/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "tsconfig/base.json", + "compilerOptions": { + "baseUrl": ".", + "types": ["node"] + } +} diff --git a/packages/utils/tsup.config.ts b/packages/utils/tsup.config.ts new file mode 100644 index 0000000..f0ac238 --- /dev/null +++ b/packages/utils/tsup.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from "tsup"; + +export default defineConfig({ + entry: ["src/index.ts"], + dts: true, + format: ["esm"], +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0b23b2f..82d8d14 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -126,6 +126,9 @@ importers: '@notation/state-sqlite': specifier: workspace:* version: link:../../packages/state-sqlite + '@notation/utils': + specifier: workspace:* + version: link:../../packages/utils devDependencies: '@types/node': specifier: ^22.13.4 @@ -352,6 +355,10 @@ importers: packages/resource: {} packages/state: + dependencies: + '@notation/utils': + specifier: workspace:* + version: link:../utils devDependencies: '@types/node': specifier: ^22.13.4 @@ -375,6 +382,9 @@ importers: '@notation/resource': specifier: workspace:* version: link:../resource + '@notation/utils': + specifier: workspace:* + version: link:../utils fflate: specifier: 0.8.2 version: 0.8.2 @@ -385,6 +395,12 @@ importers: packages/tsconfig: {} + packages/utils: + devDependencies: + '@types/node': + specifier: ^22.13.4 + version: 22.13.4 + packages: '@aws-sdk/client-apigatewayv2@3.1080.0':