History

Run credential writes inline after the winning insert

Rhys Sullivan committed 2 months ago

Commit 7c9c5fe1 parent 012a0272 files · +127 −173Browse files

Unified · Split

2 files
  1. packages/core/sdk/src/connections.test.ts +66 −94
  2. packages/core/sdk/src/executor.ts +61 −79
packages/core/sdk/src/connections.test.ts+66−94
@@ -576,17 +576,17 @@ describe("connections.create", () => { · 575 unmodified lines
576});576});
577 577
578// ---------------------------------------------------------------------------578// ---------------------------------------------------------------------------
579// Credential-write durability & compensation. The row insert and the provider579// Credential-write compensation. The row insert and the provider write cannot
580// write cannot be atomic — the provider may live outside the database — so the580// be atomic — the provider may live outside the database — so the create
581// create sequences them: the row must be DURABLE before the provider is581// sequences them: the provider is touched only after this create wins the row
582// touched, and a write that does not complete must tear down everything it582// insert, and a write that does not complete must tear down everything it
583// already stored. The worst outcome is a committed row whose credentials were583// already stored. The worst outcome is a committed row whose credentials were
584// never written: it 409s every retry while resolving nothing.584// never written: it 409s every retry while resolving nothing.
585// ---------------------------------------------------------------------------585// ---------------------------------------------------------------------------
586 586
587const trackingProvider = (587const trackingProvider = (
588 store: Map<string, string>,588 store: Map<string, string>,
589 overrides?: Partial<Pick<CredentialProvider, "set">>,589 overrides?: Partial<Pick<CredentialProvider, "set" | "delete">>,
590): CredentialProvider => ({590): CredentialProvider => ({
591 key: ProviderKey.make("memory"),591 key: ProviderKey.make("memory"),
592 writable: true,592 writable: true,
@@ -607,98 +607,10 @@ const durabilityPlugin = (provider: CredentialProvider) => · 14 unmodified lines
607 extension: (ctx) => ({607 extension: (ctx) => ({
608 seed: () =>608 seed: () =>
609 ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }),609 ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }),
610 resolveValue: (name: string) =>
611 ctx.connections.resolveValue({
612 owner: "org",
613 integration: INTEG,
614 name: ConnectionName.make(name),
615 }),
616 /** A plugin composing a create into its own atomic unit — the shape that
617 * makes the create's inner insert a pass-through with no commit of its
618 * own. */
619 createInTransaction: (rollback: boolean) =>
620 ctx.transaction(
621 Effect.gen(function* () {
622 const created = yield* ctx.connections.create({
623 owner: "org",
624 name: ConnectionName.make("main"),
625 integration: INTEG,
626 template: TEMPLATE,
627 value: "secret-token",
628 });
629 if (rollback) return yield* Effect.fail("rollback" as const);
630 return created;
631 }),
632 ),
633 }),610 }),
634 }))();611 }))();
635 612
636describe("connections.create credential-write durability", () => {613describe("connections.create credential-write compensation", () => {
637 // `transaction` nests by pass-through: inside a plugin's `ctx.transaction`
638 // the create's inner insert commits nothing — the row becomes durable only
639 // with the OUTER commit. A credential written inline in that window outlives
640 // a rollback as an orphan at a deterministic item id, where the next create
641 // of the same name silently adopts it. The write must wait for the real
642 // commit, and the caller's own typed failure must pass through untouched.
643 it.effect("discards the credential write when an enclosing transaction rolls back", () =>
644 Effect.gen(function* () {
645 const store = new Map<string, string>();
646 const executor = yield* makeTestExecutor({
647 plugins: [durabilityPlugin(trackingProvider(store))] as const,
648 });
649 yield* executor.durable.seed();
650
651 const result = yield* Effect.result(executor.durable.createInTransaction(true));
652 expect(Result.isFailure(result)).toBe(true);
653 if (!Result.isFailure(result)) return;
654 expect(result.failure).toBe("rollback");
655
656 // No zombie row and no orphaned credential survived the rollback.
657 expect(yield* executor.connections.list()).toEqual([]);
658 expect(store.size).toBe(0);
659
660 // The name is fully reusable: a committed create stores its own value.
661 yield* executor.durable.createInTransaction(false);
662 expect((yield* executor.connections.list()).length).toBe(1);
663 expect(yield* executor.durable.resolveValue("main")).toBe("secret-token");
664 }),
665 );
666
667 // Inside an enclosing transaction the create returns before its deferred
668 // credential write runs, so a write failure can no longer become the
669 // caller's typed error. What it must NOT become is a silent zombie: the
670 // deferred failure removes the committed row (and its tools) and reports
671 // the failure loudly.
672 it.effect("a deferred credential-write failure removes the committed row and logs", () =>
673 Effect.gen(function* () {
674 const store = new Map<string, string>();
675 const provider = trackingProvider(store, {
676 set: () =>
677 Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })),
678 });
679 const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const });
680 yield* executor.durable.seed();
681
682 const errors: string[] = [];
683 const capture = Logger.make<unknown, void>((options) => {
684 if (options.logLevel === "Error") {
685 errors.push(Inspectable.toStringUnknown(options.message, 0));
686 }
687 });
688 const result = yield* Effect.result(
689 executor.durable.createInTransaction(false).pipe(Effect.provide(Logger.layer([capture]))),
690 );
691
692 // The transaction committed, so the create itself reported success ...
693 expect(Result.isSuccess(result)).toBe(true);
694 // ... but the failed write tore the committed row (and its tools) down
695 // instead of leaving it to 409 every retry while resolving nothing.
696 expect(yield* executor.connections.list()).toEqual([]);
697 expect(yield* executor.tools.list()).toEqual([]);
698 expect(errors.some((line) => line.includes("failed after commit"))).toBe(true);
699 }),
700 );
701
702 // Interruption is not an error: error-channel compensation never sees it. A614 // Interruption is not an error: error-channel compensation never sees it. A
703 // create interrupted mid-write must still tear down what it already did —615 // create interrupted mid-write must still tear down what it already did —
704 // the committed row and every item that landed before the interrupt.616 // the committed row and every item that landed before the interrupt.
@@ -773,6 +685,53 @@ describe("connections.create credential-write durability", () => { · 68 unmodified lines
773 }),685 }),
774 );686 );
775 687
688 // A provider can expose `set` without `delete`. Compensation then cannot
689 // undo the items already written — that is acceptable only if it is loud:
690 // a warning must name the item that may be stranded, never a silent skip.
691 it.effect("warns about possibly stranded items when the provider has no delete", () =>
692 Effect.gen(function* () {
693 const store = new Map<string, string>();
694 const provider = trackingProvider(store, {
695 set: (id, value) =>
696 String(id).endsWith(":second")
697 ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined }))
698 : Effect.sync(() => void store.set(String(id), value)),
699 delete: undefined,
700 });
701 const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const });
702 yield* executor.durable.seed();
703
704 const warnings: string[] = [];
705 const capture = Logger.make<unknown, void>((options) => {
706 if (options.logLevel === "Warn") {
707 warnings.push(Inspectable.toStringUnknown(options.message, 0));
708 }
709 });
710 const result = yield* Effect.result(
711 executor.connections
712 .create({
713 owner: "org",
714 name: ConnectionName.make("main"),
715 integration: INTEG,
716 template: TEMPLATE,
717 values: { first: "1", second: "2" },
718 })
719 .pipe(Effect.provide(Logger.layer([capture]))),
720 );
721
722 expect(Result.isFailure(result)).toBe(true);
723 if (!Result.isFailure(result)) return;
724 expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true);
725 // The row is gone, but the first item cannot be undone without a
726 // provider delete ...
727 expect(yield* executor.connections.list()).toEqual([]);
728 expect(store.size).toBe(1);
729 // ... and the create said so, naming the item.
730 expect(warnings.some((line) => line.includes("stranded"))).toBe(true);
731 expect(warnings.some((line) => line.includes("first"))).toBe(true);
732 }),
733 );
734
776 // The compensating delete can itself fail. Swallowing that failure strands735 // The compensating delete can itself fail. Swallowing that failure strands
777 // a visible credential-less row behind an error that never mentions it. The736 // a visible credential-less row behind an error that never mentions it. The
778 // create must fail with an error that NAMES the stranded connection so an737 // create must fail with an error that NAMES the stranded connection so an
@@ -841,6 +800,19 @@ describe("connections.create credential-write durability", () => { · 62 unmodified lines
841 if (!Predicate.isTagged("StorageError")(failure)) return;800 if (!Predicate.isTagged("StorageError")(failure)) return;
842 expect(failure.message).toContain("main");801 expect(failure.message).toContain("main");
843 expect(failure.message).toContain("vercel");802 expect(failure.message).toContain("vercel");
803 // The original write failure is retained as the cause, not replaced.
804 const isStorageError = (u: unknown): u is StorageError =>
805 Predicate.isTagged("StorageError")(u);
806 expect(isStorageError(failure.cause)).toBe(true);
807 if (!isStorageError(failure.cause)) return;
808 expect(failure.cause.message).toBe("provider write refused");
809
810 // Non-vacuous: the compensating delete really did fail, so the
811 // credential-less row the error names is still there.
812 failRowDelete = false;
813 const rows = yield* executor.connections.list();
814 expect(rows.length).toBe(1);
815 expect(String(rows[0]?.name)).toBe("main");
844 }),816 }),
845 );817 );
846});818});
packages/core/sdk/src/executor.ts+61−79
@@ -19,7 +19,6 @@ import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/sch · 18 unmodified lines
19import type { AnyColumn } from "@executor-js/fumadb/schema";19import type { AnyColumn } from "@executor-js/fumadb/schema";
20import {20import {
21 StorageError,21 StorageError,
22 activeFumaDbRef,
23 afterCommit,22 afterCommit,
24 isStorageFailure,23 isStorageFailure,
25 makeFumaClient,24 makeFumaClient,
@@ -3390,10 +3389,10 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 3364 unmodified lines
3390 }3389 }
3391 let providerKey: string;3390 let providerKey: string;
3392 const itemIds: Record<string, string> = {};3391 const itemIds: Record<string, string> = {};
3393 // Pasted-value provider writes, built here but run only after the row3392 // Pasted-value provider writes, built here but run only AFTER this
3394 // insert below is DURABLE (the post-insert block owns that ordering).3393 // create wins the row insert below. Each entry carries its own undo
3395 // Each entry carries its own undo so a write that does not complete3394 // so a write that does not complete can tear down exactly the items
3396 // can tear down exactly the items it already stored.3395 // it already stored.
3397 const pastedWrites: Array<{3396 const pastedWrites: Array<{
3398 readonly itemId: ProviderItemId;3397 readonly itemId: ProviderItemId;
3399 readonly write: Effect.Effect<void, StorageFailure>;3398 readonly write: Effect.Effect<void, StorageFailure>;
@@ -3504,20 +3503,26 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 104 unmodified lines
3504 ),3503 ),
3505 );3504 );
3506 3505
3507 // Winner-only, durable-only credential write. Only the create whose3506 // Winner-only credential write: only the create whose row insert
3508 // row insert COMMITTED may touch the provider — a pasted value's item3507 // committed may touch the provider — a pasted value's item id is
3509 // id is deterministic, so a losing create would clobber the winner's3508 // deterministic, so a losing create would clobber the winner's (or a
3510 // (or a pre-existing connection's) secret. "Committed" is load-bearing:3509 // pre-existing connection's) secret. The writes run inline, straight
3511 // `transaction` nests by pass-through, so under an enclosing plugin3510 // after the transactional insert above and BEFORE the connection's
3512 // transaction the insert above is not durable yet, and a write here3511 // tools are produced below: GraphQL/MCP plugins do authenticated
3513 // would orphan the secret if that transaction rolls back. With a3512 // introspection via `getValues()` at tool-production time, so the
3514 // transaction active the writes are deferred to its real commit via3513 // credentials must exist by then or the catalog is discovered
3515 // `afterCommit`; with none they run immediately, as before.3514 // empty/incomplete and never re-discovered.
3516 //3515 //
3517 // The immediate path deliberately does NOT go through `afterCommit`'s3516 // Known limitation (pre-existing, not addressed here): `transaction`
3518 // no-transaction fallback: that fallback swallows failures by3517 // nests by pass-through, so a create running inside an enclosing
3519 // contract, and an immediate create must keep surfacing a failed3518 // plugin `ctx.transaction` writes credentials before the OUTER
3520 // write as its own typed StorageFailure.3519 // commit. If that transaction rolls back, the row vanishes with it
3520 // but the credential items survive as orphans at their deterministic
3521 // ids — inert until the next same-shaped create overwrites them.
3522 // External credential stores cannot join a database transaction, and
3523 // deferring the write past the outer commit was tried and reverted:
3524 // it broke the tool-production ordering above and could not guarantee
3525 // the deferred hook runs exactly once under interruption.
3521 if (pastedWrites.length > 0) {3526 if (pastedWrites.length > 0) {
3522 const written: ProviderItemId[] = [];3527 const written: ProviderItemId[] = [];
3523 const writeAll = Effect.gen(function* () {3528 const writeAll = Effect.gen(function* () {
@@ -3533,10 +3538,9 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 9 unmodified lines
3533 // failing every invocation with `connection_value_missing`.3538 // failing every invocation with `connection_value_missing`.
3534 // Compensation covers BOTH halves: best-effort deletes of the items3539 // Compensation covers BOTH halves: best-effort deletes of the items
3535 // already written (the earlier secrets of a partial multi-variable3540 // already written (the earlier secrets of a partial multi-variable
3536 // write), then the row itself plus any tool rows a deferred create3541 // write), then the row itself. Nothing here is silent: every failed
3537 // produced before its commit. Nothing here is silent: every failed3542 // or impossible undo is logged, and `rowRemoved` converts a
3538 // undo is logged, and `rowRemoved` lets the immediate path convert3543 // stranded row into an error that names it.
3539 // a stranded row into an error that names it.
3540 let rowRemoved = true;3544 let rowRemoved = true;
3541 const logContext = {3545 const logContext = {
3542 owner: input.owner,3546 owner: input.owner,
@@ -3545,7 +3549,16 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 2 unmodified lines
3545 };3549 };
3546 const compensate = Effect.gen(function* () {3550 const compensate = Effect.gen(function* () {
3547 for (const entry of pastedWrites) {3551 for (const entry of pastedWrites) {
3548 if (entry.remove === null || !written.includes(entry.itemId)) continue;3552 if (!written.includes(entry.itemId)) continue;
3553 if (entry.remove === null) {
3554 // A provider exposing `set` without `delete` cannot undo its
3555 // own writes; say so instead of silently skipping.
3556 yield* Effect.logWarning(
3557 "executor connection create cannot undo a credential write: the provider has no delete, so a partial credential may be stranded",
3558 { ...logContext, item: String(entry.itemId) },
3559 );
3560 continue;
3561 }
3549 yield* entry.remove.pipe(3562 yield* entry.remove.pipe(
3550 Effect.catchCause((cause) =>3563 Effect.catchCause((cause) =>
3551 Effect.logError("executor connection create failed to undo a credential write", {3564 Effect.logError("executor connection create failed to undo a credential write", {
@@ -3556,77 +3569,46 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 4 unmodified lines
3556 ),3569 ),
3557 );3570 );
3558 }3571 }
3559 const where = (b: AnyCb) =>3572 yield* core
3560 b.and(3573 .deleteMany("connection", {
3561 byOwner(input.owner)(b),
3562 b("integration", "=", String(input.integration)),
3563 b("connection", "=", String(name)),
3564 );
3565 yield* Effect.gen(function* () {
3566 yield* core.deleteMany("tool", { where });
3567 yield* core.deleteMany("definition", { where });
3568 yield* core.deleteMany("connection", {
3569 where: (b: AnyCb) =>3574 where: (b: AnyCb) =>
3570 b.and(3575 b.and(
3571 byOwner(input.owner)(b),3576 byOwner(input.owner)(b),
3572 b("integration", "=", String(input.integration)),3577 b("integration", "=", String(input.integration)),
3573 b("name", "=", String(name)),3578 b("name", "=", String(name)),
3574 ),3579 ),
3575 });3580 })
3576 }).pipe(3581 .pipe(
3577 Effect.catchCause((cause) =>3582 Effect.catchCause((cause) =>
3578 Effect.sync(() => {3583 Effect.sync(() => {
3579 rowRemoved = false;3584 rowRemoved = false;
3580 }).pipe(3585 }).pipe(
3581 Effect.andThen(3586 Effect.andThen(
3582 Effect.logError(3587 Effect.logError(
3583 "executor connection create stranded a credential-less connection row",3588 "executor connection create stranded a credential-less connection row",
3584 { ...logContext, cause },3589 { ...logContext, cause },
3590 ),
3585 ),3591 ),
3586 ),3592 ),
3587 ),3593 ),
3588 ),3594 );
3589 );
3590 });3595 });
3591 3596
3592 // `onExit`, not `tapError`: compensation must also run when the3597 // `onExit`, not `tapError`: compensation must also run when the
3593 // write is interrupted or dies with a defect.3598 // write is interrupted or dies with a defect.
3594 const guardedWrites = writeAll.pipe(3599 yield* writeAll.pipe(
3595 Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)),3600 Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)),
3596 );3601 Effect.catch((error) =>
3597 3602 rowRemoved
3598 const enclosingTransaction = yield* Effect.service(activeFumaDbRef);3603 ? Effect.fail(error)
3599 if (enclosingTransaction === null) {3604 : Effect.fail(
3600 yield* guardedWrites.pipe(3605 new StorageError({
3601 Effect.catch((error) =>3606 message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded without stored credentials and must be removed manually.`,
3602 rowRemoved3607 cause: error,
3603 ? Effect.fail(error)3608 }),
3604 : Effect.fail(
3605 new StorageError({
3606 message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded without stored credentials and must be removed manually.`,
3607 cause: error,
3608 }),
3609 ),
3610 ),
3611 );
3612 } else {
3613 // The create returns to its enclosing transaction before this
3614 // hook runs, so a failure here can no longer become the caller's
3615 // typed error — compensation plus a loud log IS the contract.
3616 // Between the outer commit and this hook the row is briefly
3617 // visible without its credentials; closing that window takes a
3618 // reservation/staging scheme and is deliberately out of scope.
3619 yield* afterCommit(
3620 guardedWrites.pipe(
3621 Effect.catchCause((cause) =>
3622 Effect.logError(
3623 "executor connection create credential write failed after commit",
3624 { ...logContext, rowRemoved, cause },
3625 ),3609 ),
3626 ),3610 ),
3627 ),3611 );
3628 );
3629 }
3630 }3612 }
3631 3613
3632 // Record the sighting. The request seam (`makeScopedExecutor`) already3614 // Record the sighting. The request seam (`makeScopedExecutor`) already