History

Guard connection-create compensation by inserted row identity

Rhys Sullivan committed 2 months ago

Commit a9881951 parent 7c9c5fe3 files · +348 −79Browse files

Unified · Split

3 files
  1. .changeset/connection-create-conflict.md +1 −1
  2. packages/core/sdk/src/connections.test.ts +233 −34
  3. packages/core/sdk/src/executor.ts +114 −44
.changeset/connection-create-conflict.md+1−1
@@ -6,6 +6,6 @@ · 5 unmodified lines
66
77 `connections.create` used to upsert: a create with the same (owner, integration, name) replaced the saved connection and, for a pasted value, overwrote the stored secret itself. It now fails with the new `ConnectionAlreadyExistsError` and leaves the existing connection untouched. Remove the connection first, or pick a different name.
88
9−This adds one error to the wire contract: the `POST /connections` endpoint can answer **HTTP 409** with tag `ConnectionAlreadyExistsError`, and the `connections.create` core tool resolves the same case as `{ ok: false, error: { code: "connection_already_exists" } }`. The change is additive — no existing status, field, or success shape moves.
9+This adds one error to the wire contract: the `POST /connections` endpoint can answer **HTTP 409** with tag `ConnectionAlreadyExistsError`, and the `connections.create` core tool resolves the same case as `{ ok: false, error: { code: "connection_already_exists" } }`. The core tool now also resolves the other expected input failures the same way instead of as opaque internal errors: `integration_not_found` for an unknown integration and `invalid_connection_input` for an invalid input. The change is additive — no existing status, field, or success shape moves.
1010
1111 OAuth is unaffected. Fresh OAuth connects already resolve a taken name to the next free suffix through `newConnection`, and reconnect still re-mints the same connection on purpose.
packages/core/sdk/src/connections.test.ts+233−34
@@ -1,7 +1,9 @@
11 import { describe, expect, it } from "@effect/vitest";
22 import {
3+ Cause,
34 Deferred,
45 Effect,
6+ Exit,
57 Fiber,
68 Inspectable,
79 Logger,
@@ -610,6 +612,37 @@ const durabilityPlugin = (provider: CredentialProvider) => · 602 unmodified lines
610612 }),
611613 }))();
612614
615+/** Wrap a test `FumaDb` so deletes on the `connection` table can be made to
616+ * fail on demand — the raw driver-level failure the compensating delete must
617+ * survive loudly. Transactions hand out wrapped handles too, so the guarded
618+ * delete inside the compensation transaction is covered. */
619+const failableConnectionDeletes = (db: FumaDb, shouldFail: () => boolean): FumaDb => {
620+ const wrap = (inner: FumaDb): FumaDb =>
621+ new Proxy(inner, {
622+ get(target, prop) {
623+ if (prop === "withContext") {
624+ return (context: unknown) =>
625+ wrap((target.withContext as (c: unknown) => FumaDb)(context));
626+ }
627+ if (prop === "transaction") {
628+ return (run: (tx: FumaDb) => Promise<unknown>) =>
629+ (target.transaction as (r: (tx: FumaDb) => Promise<unknown>) => Promise<unknown>)(
630+ (tx) => run(wrap(tx)),
631+ );
632+ }
633+ if (prop === "deleteMany") {
634+ return (table: unknown, query: unknown) =>
635+ shouldFail() && table === "connection"
636+ ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle
637+ Promise.reject(new StorageError({ message: "delete refused", cause: undefined }))
638+ : (target.deleteMany as (t: unknown, q: unknown) => Promise<unknown>)(table, query);
639+ }
640+ return Reflect.get(target, prop);
641+ },
642+ });
643+ return wrap(db);
644+};
645+
613646 describe("connections.create credential-write compensation", () => {
614647 // Interruption is not an error: error-channel compensation never sees it. A
615648 // create interrupted mid-write must still tear down what it already did —
@@ -739,38 +772,6 @@ describe("connections.create credential-write compensation", () => { · 123 unmodified lines
739772 it.effect("names the stranded connection when the compensating delete fails", () =>
740773 Effect.gen(function* () {
741774 let failRowDelete = false;
742− const failConnectionDeletes = (db: FumaDb): FumaDb => {
743− const wrap = (inner: FumaDb): FumaDb =>
744− new Proxy(inner, {
745− get(target, prop) {
746− if (prop === "withContext") {
747− return (context: unknown) =>
748− wrap((target.withContext as (c: unknown) => FumaDb)(context));
749− }
750− if (prop === "transaction") {
751− return (run: (tx: FumaDb) => Promise<unknown>) =>
752− (target.transaction as (r: (tx: FumaDb) => Promise<unknown>) => Promise<unknown>)(
753− (tx) => run(wrap(tx)),
754− );
755− }
756− if (prop === "deleteMany") {
757− return (table: unknown, query: unknown) =>
758− failRowDelete && table === "connection"
759− ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle
760− Promise.reject(
761− new StorageError({ message: "delete refused", cause: undefined }),
762− )
763− : (target.deleteMany as (t: unknown, q: unknown) => Promise<unknown>)(
764− table,
765− query,
766− );
767− }
768− return Reflect.get(target, prop);
769− },
770− });
771− return wrap(db);
772− };
773−
774775 const store = new Map<string, string>();
775776 const provider = trackingProvider(store, {
776777 set: (id, value) =>
@@ -779,7 +780,10 @@ describe("connections.create credential-write compensation", () => { · 2 unmodified lines
779780 : Effect.sync(() => void store.set(String(id), value)),
780781 });
781782 const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const });
782− const executor = yield* createExecutor({ ...config, db: failConnectionDeletes(config.db) });
783+ const executor = yield* createExecutor({
784+ ...config,
785+ db: failableConnectionDeletes(config.db, () => failRowDelete),
786+ });
783787 yield* executor.durable.seed();
784788 failRowDelete = true;
785789
@@ -808,13 +812,208 @@ describe("connections.create credential-write compensation", () => { · 22 unmodified lines
808812 expect(failure.cause.message).toBe("provider write refused");
809813
810814 // Non-vacuous: the compensating delete really did fail, so the
811− // credential-less row the error names is still there.
815+ // row the error names is still there.
816+ failRowDelete = false;
817+ const rows = yield* executor.connections.list();
818+ expect(rows.length).toBe(1);
819+ expect(String(rows[0]?.name)).toBe("main");
820+ }),
821+ );
822+
823+ // Compensation can be slow (provider calls). In that window a concurrent
824+ // remove can free the name and a new create can take it, writing fresh
825+ // secrets at the SAME deterministic item ids. Late compensation must then
826+ // recognize that the row is no longer the one it inserted — identified by
827+ // the storage surrogate row id — and touch neither the replacement row nor
828+ // its credentials. Losing compensation to a concurrent remove is correct:
829+ // the remover already cleaned up.
830+ it.effect("late compensation leaves a concurrent replacement untouched", () =>
831+ Effect.scoped(
832+ Effect.gen(function* () {
833+ const secondWriteEntered = yield* Deferred.make<void>();
834+ const releaseSecondWrite = yield* Deferred.make<void>();
835+ const store = new Map<string, string>();
836+ let parkNextSecondWrite = true;
837+ const provider = trackingProvider(store, {
838+ set: (id, value) => {
839+ if (String(id).endsWith(":second") && parkNextSecondWrite) {
840+ parkNextSecondWrite = false;
841+ return Deferred.succeed(secondWriteEntered, undefined).pipe(
842+ Effect.andThen(Deferred.await(releaseSecondWrite)),
843+ Effect.andThen(
844+ Effect.fail(
845+ new StorageError({ message: "provider write refused", cause: undefined }),
846+ ),
847+ ),
848+ );
849+ }
850+ return Effect.sync(() => void store.set(String(id), value));
851+ },
852+ });
853+ const executor = yield* makeTestExecutor({
854+ plugins: [durabilityPlugin(provider)] as const,
855+ });
856+ yield* executor.durable.seed();
857+
858+ const fiber = yield* Effect.forkChild(
859+ executor.connections.create({
860+ owner: "org",
861+ name: ConnectionName.make("main"),
862+ integration: INTEG,
863+ template: TEMPLATE,
864+ values: { first: "a-1", second: "a-2" },
865+ }),
866+ );
867+ yield* Deferred.await(secondWriteEntered);
868+
869+ // While the first create is parked in its provider write, the user
870+ // removes the connection and recreates it with different secrets.
871+ yield* executor.connections.remove({
872+ owner: "org",
873+ integration: INTEG,
874+ name: ConnectionName.make("main"),
875+ });
876+ yield* executor.connections.create({
877+ owner: "org",
878+ name: ConnectionName.make("main"),
879+ integration: INTEG,
880+ template: TEMPLATE,
881+ values: { first: "c-1", second: "c-2" },
882+ });
883+
884+ // Release the parked write: the first create fails and compensates
885+ // late, against a name it no longer owns.
886+ yield* Deferred.succeed(releaseSecondWrite, undefined);
887+ const exit = yield* Fiber.await(fiber);
888+ expect(Exit.isFailure(exit)).toBe(true);
889+
890+ // The replacement row AND its credentials survive.
891+ const rows = yield* executor.connections.list();
892+ expect(rows.length).toBe(1);
893+ expect(String(rows[0]?.name)).toBe("main");
894+ expect(store.get("connection:org:vercel:main:first")).toBe("c-1");
895+ expect(store.get("connection:org:vercel:main:second")).toBe("c-2");
896+ }),
897+ ),
898+ );
899+
900+ // A provider write can die with a defect instead of failing. The stranded-
901+ // row promise must hold there too: a defect followed by a failed
902+ // compensating delete surfaces the same typed StorageError naming the
903+ // stranded connection, not an anonymous crash.
904+ it.effect("a defect followed by a failed row delete still names the stranded connection", () =>
905+ Effect.gen(function* () {
906+ let failRowDelete = false;
907+ const store = new Map<string, string>();
908+ const provider = trackingProvider(store, {
909+ set: (id, value) =>
910+ String(id).endsWith(":second")
911+ ? Effect.die("provider crashed")
912+ : Effect.sync(() => void store.set(String(id), value)),
913+ });
914+ const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const });
915+ const executor = yield* createExecutor({
916+ ...config,
917+ db: failableConnectionDeletes(config.db, () => failRowDelete),
918+ });
919+ yield* executor.durable.seed();
920+ failRowDelete = true;
921+
922+ const exit = yield* Effect.exit(
923+ executor.connections.create({
924+ owner: "org",
925+ name: ConnectionName.make("main"),
926+ integration: INTEG,
927+ template: TEMPLATE,
928+ values: { first: "1", second: "2" },
929+ }),
930+ );
931+
932+ expect(Exit.isFailure(exit)).toBe(true);
933+ if (!Exit.isFailure(exit)) return;
934+ // Not an anonymous defect: the typed error is on the failure channel.
935+ expect(Cause.hasFails(exit.cause)).toBe(true);
936+ const failure = Cause.squash(exit.cause);
937+ const isStorageError = (u: unknown): u is StorageError =>
938+ Predicate.isTagged("StorageError")(u);
939+ expect(isStorageError(failure)).toBe(true);
940+ if (!isStorageError(failure)) return;
941+ expect(failure.message).toContain("main");
942+ expect(failure.message).toContain("vercel");
943+ expect(failure.message).toContain("stranded");
944+ // The original defect is retained as the cause, not replaced.
945+ expect(failure.cause).toBe("provider crashed");
946+
947+ // Non-vacuous: the row the error names is still there.
812948 failRowDelete = false;
813949 const rows = yield* executor.connections.list();
814950 expect(rows.length).toBe(1);
815951 expect(String(rows[0]?.name)).toBe("main");
816952 }),
817953 );
954+
955+ // Interruption cannot carry a typed error — interrupting wins over failing
956+ // — so when an interrupted create cannot delete its row, the stranded row
957+ // is reported through a loud error log and the create stays an
958+ // interruption. The items that landed before the interrupt stay with the
959+ // stranded row: credential teardown is gated on the row delete succeeding.
960+ it.effect("an interrupted create with a failed row delete logs the stranded row", () =>
961+ Effect.scoped(
962+ Effect.gen(function* () {
963+ let failRowDelete = false;
964+ const secondWriteEntered = yield* Deferred.make<void>();
965+ const store = new Map<string, string>();
966+ const provider = trackingProvider(store, {
967+ set: (id, value) =>
968+ String(id).endsWith(":second")
969+ ? Deferred.succeed(secondWriteEntered, undefined).pipe(Effect.andThen(Effect.never))
970+ : Effect.sync(() => void store.set(String(id), value)),
971+ });
972+ const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const });
973+ const executor = yield* createExecutor({
974+ ...config,
975+ db: failableConnectionDeletes(config.db, () => failRowDelete),
976+ });
977+ yield* executor.durable.seed();
978+
979+ const errors: string[] = [];
980+ const capture = Logger.make<unknown, void>((options) => {
981+ if (options.logLevel === "Error") {
982+ errors.push(Inspectable.toStringUnknown(options.message, 0));
983+ }
984+ });
985+ const fiber = yield* Effect.forkChild(
986+ executor.connections
987+ .create({
988+ owner: "org",
989+ name: ConnectionName.make("main"),
990+ integration: INTEG,
991+ template: TEMPLATE,
992+ values: { first: "1", second: "2" },
993+ })
994+ .pipe(Effect.provide(Logger.layer([capture]))),
995+ );
996+ yield* Deferred.await(secondWriteEntered);
997+ failRowDelete = true;
998+ yield* Fiber.interrupt(fiber);
999+ const exit = yield* Fiber.await(fiber);
1000+
1001+ // Still an interruption — and the stranded row was reported loudly.
1002+ expect(Exit.isFailure(exit)).toBe(true);
1003+ if (!Exit.isFailure(exit)) return;
1004+ expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true);
1005+ expect(errors.some((line) => line.includes("stranded"))).toBe(true);
1006+
1007+ // Non-vacuous: the row survived the failed delete, and the item that
1008+ // landed before the interrupt stayed with it.
1009+ failRowDelete = false;
1010+ const rows = yield* executor.connections.list();
1011+ expect(rows.length).toBe(1);
1012+ expect(String(rows[0]?.name)).toBe("main");
1013+ expect(store.size).toBe(1);
1014+ }),
1015+ ),
1016+ );
8181017 });
8191018
8201019 describe("connections.list / get", () => {
packages/core/sdk/src/executor.ts+114−44
@@ -1,4 +1,5 @@
11 import {
2+ Cause,
23 Deferred,
34 Duration,
45 Effect,
@@ -8,6 +9,7 @@ import { · 3 unmodified lines
89 Layer,
910 Option,
1011 Predicate,
12+ Ref,
1113 Schema,
1214 Semaphore,
1315 } from "effect";
@@ -3454,7 +3456,19 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 3440 unmodified lines
34543456 catch: (cause) => storageFailureFromUnknown("invalid owner", cause),
34553457 });
34563458 const now = new Date();
3457− yield* transaction(
3459+ // The storage surrogate id of the row THIS create inserted. The
3460+ // composite key (owner, integration, name) can change hands while a
3461+ // failed create is still compensating, and `created_at` round-trips
3462+ // at second precision, so neither identifies OUR row — only `row_id`
3463+ // does. `FumaRow` deliberately hides `row_id` from domain rows, so it
3464+ // is read through a narrow cast. Every adapter generates it ORM-side
3465+ // on insert; a create result without it is a broken storage contract
3466+ // and fails here, inside the transaction, before any provider write.
3467+ const rowIdOf = (row: unknown): string | null => {
3468+ const value = row == null ? null : (row as Record<string, unknown>)["row_id"];
3469+ return typeof value === "string" ? value : null;
3470+ };
3471+ const insertedRowId = yield* transaction(
34583472 Effect.gen(function* () {
34593473 const existing = yield* findConnectionRow({
34603474 owner: input.owner,
@@ -3468,7 +3482,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 7 unmodified lines
34683482 name,
34693483 });
34703484 }
3471− yield* core.create("connection", {
3485+ const inserted = yield* core.create("connection", {
34723486 tenant: keys.tenant,
34733487 owner: keys.owner,
34743488 subject: keys.subject,
@@ -3487,6 +3501,15 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 12 unmodified lines
34873501 created_at: now,
34883502 updated_at: now,
34893503 });
3504+ const rowId = rowIdOf(inserted);
3505+ if (rowId === null) {
3506+ return yield* new StorageError({
3507+ message:
3508+ "Storage adapter did not return the inserted connection row's row_id; the create cannot be compensated safely.",
3509+ cause: undefined,
3510+ });
3511+ }
3512+ return rowId;
34903513 }),
34913514 ).pipe(
34923515 // Both racers can observe absence and reach the insert; the primary
@@ -3535,19 +3558,82 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 42 unmodified lines
35353558 // While the committed row exists no concurrent create can win, so
35363559 // on an incomplete write it is ours to tear down — a surviving row
35373560 // whose item_ids were never stored would 409 every retry while
3538− // failing every invocation with `connection_value_missing`.
3539− // Compensation covers BOTH halves: best-effort deletes of the items
3540− // already written (the earlier secrets of a partial multi-variable
3541− // write), then the row itself. Nothing here is silent: every failed
3542− // or impossible undo is logged, and `rowRemoved` converts a
3543− // stranded row into an error that names it.
3544− let rowRemoved = true;
3561+ // failing every invocation with `connection_value_missing`. But
3562+ // "ours" needs proof before anything is deleted: the composite key
3563+ // (owner, integration, name) can change hands while compensation is
3564+ // still pending (provider calls can be slow) — a concurrent remove
3565+ // frees the name, a new create takes it and writes fresh secrets at
3566+ // the SAME deterministic item ids. The one column that tells our
3567+ // row apart from such a successor is `insertedRowId`, so the row
3568+ // delete carries it in its WHERE (guarded delete), and the identity
3569+ // check runs in the same transaction as the delete so both see one
3570+ // consistent row.
3571+ //
3572+ // Order matters: the ROW is deleted first, and the credential items
3573+ // are undone only when the guarded delete actually removed OUR row.
3574+ // If the row is already gone or replaced, losing compensation is
3575+ // correct — the remover already cleaned up, and the deterministic
3576+ // item ids may by now carry the successor's secrets, so deleting
3577+ // them here would clobber a healthy connection. Nothing here is
3578+ // silent: every failed or impossible undo is logged, and
3579+ // `rowOutcome` converts a stranded row into an error that names it.
3580+ const rowOutcomeRef = yield* Ref.make<"removed" | "superseded" | "failed">("removed");
35453581 const logContext = {
35463582 owner: input.owner,
35473583 integration: String(input.integration),
35483584 connection: String(name),
35493585 };
35503586 const compensate = Effect.gen(function* () {
3587+ const rowOutcome = yield* transaction(
3588+ Effect.gen(function* () {
3589+ const current = yield* findConnectionRow({
3590+ owner: input.owner,
3591+ integration: input.integration,
3592+ name,
3593+ });
3594+ if (rowIdOf(current) !== insertedRowId) {
3595+ return "superseded" as const;
3596+ }
3597+ yield* core.deleteMany("connection", {
3598+ where: (b: AnyCb) =>
3599+ b.and(
3600+ byOwner(input.owner)(b),
3601+ b("integration", "=", String(input.integration)),
3602+ b("name", "=", String(name)),
3603+ // Even if the row changed hands between the read above
3604+ // and this statement, only OUR row can match.
3605+ b("row_id", "=", insertedRowId),
3606+ ),
3607+ });
3608+ return "removed" as const;
3609+ }),
3610+ ).pipe(
3611+ Effect.catchCause((cause) =>
3612+ Effect.logError(
3613+ "executor connection create stranded a connection row it could not delete",
3614+ { ...logContext, cause },
3615+ ).pipe(Effect.as("failed" as const)),
3616+ ),
3617+ );
3618+ yield* Ref.set(rowOutcomeRef, rowOutcome);
3619+ if (rowOutcome === "superseded") {
3620+ // A concurrent remove took our row, and a successor may already
3621+ // own the name and the item ids. The remover cleaned up;
3622+ // nothing left here is ours to touch.
3623+ yield* Effect.logInfo(
3624+ "executor connection create skipped compensation: the connection row was already removed or replaced",
3625+ logContext,
3626+ );
3627+ return;
3628+ }
3629+ if (rowOutcome === "failed") {
3630+ // The row delete failed, so the row — still ours — keeps
3631+ // holding the name together with the items that already
3632+ // landed. Leave the items in place (they belong to the
3633+ // stranded row the caller is told to remove) and let the exit
3634+ // handling below surface the error.
3635+ return;
3636+ }
35513637 for (const entry of pastedWrites) {
35523638 if (!written.includes(entry.itemId)) continue;
35533639 if (entry.remove === null) {
@@ -3569,46 +3655,30 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 15 unmodified lines
35693655 ),
35703656 );
35713657 }
3572− yield* core
3573− .deleteMany("connection", {
3574− where: (b: AnyCb) =>
3575− b.and(
3576− byOwner(input.owner)(b),
3577− b("integration", "=", String(input.integration)),
3578− b("name", "=", String(name)),
3579− ),
3580− })
3581− .pipe(
3582− Effect.catchCause((cause) =>
3583− Effect.sync(() => {
3584− rowRemoved = false;
3585− }).pipe(
3586− Effect.andThen(
3587− Effect.logError(
3588− "executor connection create stranded a credential-less connection row",
3589− { ...logContext, cause },
3590− ),
3591− ),
3592− ),
3593− ),
3594− );
35953658 });
35963659
35973660 // `onExit`, not `tapError`: compensation must also run when the
3598− // write is interrupted or dies with a defect.
3599− yield* writeAll.pipe(
3661+ // write is interrupted or dies with a defect. The stranded-row
3662+ // promise must hold on every one of those exit shapes, so the exit
3663+ // is captured and re-raised by hand: a typed failure or a defect
3664+ // that left the row behind becomes the StorageError below, while an
3665+ // interruption cannot carry a typed error at all (interrupting wins
3666+ // over failing) — for it the loud log inside `compensate` is the
3667+ // only signal, and the interruption is re-raised untouched.
3668+ const writeExit = yield* writeAll.pipe(
36003669 Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)),
3601− Effect.catch((error) =>
3602− rowRemoved
3603− ? Effect.fail(error)
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− ),
3670+ Effect.exit,
36113671 );
3672+ if (Exit.isFailure(writeExit)) {
3673+ const rowOutcome = yield* Ref.get(rowOutcomeRef);
3674+ if (rowOutcome === "failed" && !Cause.hasInterruptsOnly(writeExit.cause)) {
3675+ return yield* new StorageError({
3676+ 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 with incomplete credentials and must be removed manually.`,
3677+ cause: Cause.squash(writeExit.cause),
3678+ });
3679+ }
3680+ return yield* Effect.failCause(writeExit.cause);
3681+ }
36123682 }
36133683
36143684 // Record the sighting. The request seam (`makeScopedExecutor`) already

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t