refactor(db): rename claimOne adapter primitive to consumeOne (#9568)

This commit is contained in:
Gustavo Valverde
2026-05-12 07:44:07 +00:00
committed by GitHub
parent 62b8793c11
commit 0cbddb8fa4
17 changed files with 74 additions and 69 deletions
@@ -9,4 +9,4 @@
"better-auth": patch
---
Add `internalAdapter.consumeVerificationValue(identifier)`: atomically consume a verification row keyed by identifier. The first concurrent caller receives the row; later racers receive `null`. Backed by a new `DBAdapter.claimOne` primitive implemented natively per adapter (memory, mongo, drizzle, kysely, prisma), with a `transaction(findMany + delete)` factory fallback. `SecondaryStorage.getAndDelete` is added as an optional companion; Redis ships it via an atomic Lua get-and-delete operation for compatibility with Redis versions before 6.2.
Add `internalAdapter.consumeVerificationValue(identifier)`: atomically consume a verification row keyed by identifier. The first concurrent caller receives the row; later racers receive `null`. Backed by a new `DBAdapter.consumeOne` primitive implemented natively per adapter (memory, mongo, drizzle, kysely, prisma), with a `transaction(findMany + delete)` factory fallback. `SecondaryStorage.getAndDelete` is added as an optional companion; Redis ships it via an atomic Lua get-and-delete operation for compatibility with Redis versions before 6.2.
@@ -1315,23 +1315,23 @@ export const createInternalAdapter = (
// FIXME(consume-identifier-atomic): add an adapter primitive that
// deletes all rows for an identifier and returns the latest row in
// one operation. Until then, claim the latest row as the race gate
// one operation. Until then, consume the latest row as the race gate
// and invalidate stale rows inside the same transaction/local lock.
const hookWhere = [{ field: "id", value: latest.id }];
return consumeOneWithHooks<Verification>(
"verification",
hookWhere,
async () => {
const claimed = await txAdapter.claimOne<Verification>({
const consumed = await txAdapter.consumeOne<Verification>({
model: "verification",
where: hookWhere,
});
if (!claimed) return null;
if (!consumed) return null;
await txAdapter.deleteMany({
model: "verification",
where,
});
return claimed;
return consumed;
},
latest,
);
+11 -11
View File
@@ -416,25 +416,25 @@ export function getWithHooks(
}
/**
* Wraps an atomic claim operation in the plugin `delete.before` and
* `delete.after` hook lifecycle. The caller supplies a `claimFn` that
* Wraps an atomic consume operation in the plugin `delete.before` and
* `delete.after` hook lifecycle. The caller supplies a `consumeFn` that
* performs the actual single-row delete-and-return (typically the
* adapter's `claimOne`). The first concurrent caller wins, subsequent
* adapter's `consumeOne`). The first concurrent caller wins, subsequent
* racers resolve to `null` without firing `delete.after` hooks.
*
* `preSnapshot` lets the caller hand in a row it already fetched so
* `delete.before` hooks don't trigger a second read. Without it, the
* helper falls back to a best-effort `findMany` against `hookWhere`.
* The snapshot only feeds `delete.before`; the `claimFn` return value
* The snapshot only feeds `delete.before`; the `consumeFn` return value
* is the race gate.
*
* Returning `false` from a `delete.before` hook aborts the claim and
* the helper resolves to `null` (no `claimFn` call, no after hooks).
* Returning `false` from a `delete.before` hook aborts the consume and
* the helper resolves to `null` (no `consumeFn` call, no after hooks).
*/
async function consumeOneWithHooks<T extends Record<string, any>>(
model: BaseModelNames,
hookWhere: Where[],
claimFn: () => Promise<T | null>,
consumeFn: () => Promise<T | null>,
preSnapshot?: T | null,
): Promise<T | null> {
const context = await getCurrentAuthContext().catch(() => null);
@@ -476,8 +476,8 @@ export function getWithHooks(
}
}
const claimed = await claimFn();
if (!claimed) return null;
const consumed = await consumeFn();
if (!consumed) return null;
for (const { source, hooks } of hooksEntries) {
const toRun = hooks[model]?.delete?.after;
@@ -492,13 +492,13 @@ export function getWithHooks(
},
() =>
// @ts-expect-error context type mismatch
toRun(claimed as any, context),
toRun(consumed as any, context),
);
});
}
}
return claimed;
return consumed;
}
return {
+1 -1
View File
@@ -61,7 +61,7 @@ function createMockAdapter(adapterId: string, dialect?: string): DBAdapter {
deleteMany: async () => {
throw new Error("Mock adapter methods should not be called");
},
claimOne: async () => {
consumeOne: async () => {
throw new Error("Mock adapter methods should not be called");
},
transaction: async (callback) => {
+2 -2
View File
@@ -1093,7 +1093,7 @@ describe("--adapter flag support (mock adapter)", () => {
deleteMany: async () => {
throw new Error("Mock adapter methods should not be called");
},
claimOne: async () => {
consumeOne: async () => {
throw new Error("Mock adapter methods should not be called");
},
transaction: async (callback) => {
@@ -1335,7 +1335,7 @@ describe("--dialect flag support", () => {
deleteMany: async () => {
throw new Error("Mock adapter methods should not be called");
},
claimOne: async () => {
consumeOne: async () => {
throw new Error("Mock adapter methods should not be called");
},
transaction: async (callback) => {
+9 -7
View File
@@ -56,7 +56,7 @@ function createTestAdapter({
});
}
describe("createAdapterFactory claimOne fallback", () => {
describe("createAdapterFactory consumeOne fallback", () => {
it("uses transaction adapter methods without double-transforming input or output", async () => {
const findMany: CustomAdapter["findMany"] = async <T>(
params: Parameters<CustomAdapter["findMany"]>[0],
@@ -111,16 +111,18 @@ describe("createAdapterFactory claimOne fallback", () => {
}),
});
const result = await adapter.claimOne<{ id: string; identifier: string }>({
model: "verification",
where: [{ field: "identifier", value: "token" }],
});
const result = await adapter.consumeOne<{ id: string; identifier: string }>(
{
model: "verification",
where: [{ field: "identifier", value: "token" }],
},
);
expect(result?.id).toBe("verification-id");
expect(result?.identifier).toBe("stored-token:output");
});
it("returns null when the delete loses the claim race", async () => {
it("returns null when the delete loses the consume race", async () => {
const adapter = createTestAdapter({
adapter: createCustomAdapter({
findMany: async <T>() =>
@@ -134,7 +136,7 @@ describe("createAdapterFactory claimOne fallback", () => {
}),
});
const result = await adapter.claimOne({
const result = await adapter.consumeOne({
model: "verification",
where: [{ field: "identifier", value: "token" }],
});
+23 -20
View File
@@ -133,7 +133,10 @@ export const createAdapterFactory =
!config.debugLogs.deleteMany
) {
return;
} else if (method === "claimOne" && !config.debugLogs.claimOne) {
} else if (
method === "consumeOne" &&
!config.debugLogs.consumeOne
) {
return;
} else if (method === "count" && !config.debugLogs.count) {
return;
@@ -487,7 +490,7 @@ export const createAdapterFactory =
| "updateMany"
| "delete"
| "deleteMany"
| "claimOne"
| "consumeOne"
| "count";
}): W extends undefined ? undefined : CleanedWhere[] => {
if (!where) return undefined as any;
@@ -1315,7 +1318,7 @@ export const createAdapterFactory =
);
return res;
},
claimOne: async <T>({
consumeOne: async <T>({
model: unsafeModel,
where: unsafeWhere,
}: {
@@ -1328,42 +1331,42 @@ export const createAdapterFactory =
const where = transformWhereClause({
model: unsafeModel,
where: unsafeWhere,
action: "claimOne",
action: "consumeOne",
});
unsafeModel = getDefaultModelName(unsafeModel);
debugLog(
{ method: "claimOne" },
{ method: "consumeOne" },
`${formatTransactionId(thisTransactionId)} ${formatStep(1, 3)}`,
`${formatMethod("claimOne")} ${formatAction("ClaimOne")}:`,
`${formatMethod("consumeOne")} ${formatAction("ConsumeOne")}:`,
{ model, where },
);
let res: T | null;
let resultNeedsOutputTransform = true;
if (adapterInstance.claimOne) {
if (adapterInstance.consumeOne) {
res = await withSpan(
`db claimOne ${model}`,
`db consumeOne ${model}`,
{
[ATTR_DB_OPERATION_NAME]: "claimOne",
[ATTR_DB_OPERATION_NAME]: "consumeOne",
[ATTR_DB_COLLECTION_NAME]: model,
},
() => adapterInstance.claimOne!<T>({ model, where }),
() => adapterInstance.consumeOne!<T>({ model, where }),
);
} else {
// TODO(claim-one-required): adapters without native `claimOne`
// TODO(consume-one-required): adapters without native `consumeOne`
// fall back to `transaction(findMany + deleteMany)`. Race-safe on
// engines with real transaction isolation; race window narrows
// (does not close) on adapters that fall through to sequential
// execution. Remove this branch when claimOne becomes required.
// FIXME(claim-one-nested-transaction): custom adapters without a
// native claimOne have no portable signal for "already inside a
// execution. Remove this branch when consumeOne becomes required.
// FIXME(consume-one-nested-transaction): custom adapters without a
// native consumeOne have no portable signal for "already inside a
// transaction". First-party adapters mark transaction-scoped
// adapters as as-is; make that capability explicit in the next
// breaking adapter contract.
res = await withSpan(
`db claimOne ${model}`,
`db consumeOne ${model}`,
{
[ATTR_DB_OPERATION_NAME]: "claimOne",
[ATTR_DB_OPERATION_NAME]: "consumeOne",
[ATTR_DB_COLLECTION_NAME]: model,
},
() =>
@@ -1395,9 +1398,9 @@ export const createAdapterFactory =
}
debugLog(
{ method: "claimOne" },
{ method: "consumeOne" },
`${formatTransactionId(thisTransactionId)} ${formatStep(2, 3)}`,
`${formatMethod("claimOne")} ${formatAction("DB Result")}:`,
`${formatMethod("consumeOne")} ${formatAction("DB Result")}:`,
{ model, data: res },
);
let transformed: any = res;
@@ -1414,9 +1417,9 @@ export const createAdapterFactory =
);
}
debugLog(
{ method: "claimOne" },
{ method: "consumeOne" },
`${formatTransactionId(thisTransactionId)} ${formatStep(3, 3)}`,
`${formatMethod("claimOne")} ${formatAction("Parsed Result")}:`,
`${formatMethod("consumeOne")} ${formatAction("Parsed Result")}:`,
{ model, data: transformed },
);
return transformed as T | null;
+8 -8
View File
@@ -15,7 +15,7 @@ export type DBAdapterDebugLogOption =
findMany?: boolean | undefined;
delete?: boolean | undefined;
deleteMany?: boolean | undefined;
claimOne?: boolean | undefined;
consumeOne?: boolean | undefined;
count?: boolean | undefined;
}
| {
@@ -212,7 +212,7 @@ export interface DBAdapterFactoryConfig<
| "updateMany"
| "delete"
| "deleteMany"
| "claimOne"
| "consumeOne"
| "count";
/**
* The model name.
@@ -448,7 +448,7 @@ export type DBAdapter<Options extends BetterAuthOptions = BetterAuthOptions> = {
delete: <_T>(data: { model: string; where: Where[] }) => Promise<void>;
deleteMany: (data: { model: string; where: Where[] }) => Promise<number>;
/**
* Atomically claim a single row matching the where clause: delete it and
* Atomically consume a single row matching the where clause: delete it and
* return the deleted row, or return `null` if no row matched.
* Implementations MUST NOT delete any additional rows that also match a
* non-unique predicate.
@@ -459,11 +459,11 @@ export type DBAdapter<Options extends BetterAuthOptions = BetterAuthOptions> = {
* (verification tokens, authorization codes, one-time tokens).
*
* Always defined on the factory-wrapped adapter. When the underlying
* `CustomAdapter` does not implement `claimOne`, the factory provides
* `CustomAdapter` does not implement `consumeOne`, the factory provides
* a fallback that wraps `findMany + deleteMany` in `transaction(...)`
* and returns the row only when the delete reports an affected row.
*/
claimOne: <T>(data: { model: string; where: Where[] }) => Promise<T | null>;
consumeOne: <T>(data: { model: string; where: Where[] }) => Promise<T | null>;
/**
* Execute multiple operations in a transaction.
* If the adapter doesn't support transactions, operations will be executed sequentially.
@@ -551,15 +551,15 @@ export interface CustomAdapter {
where: CleanedWhere[];
}) => Promise<number>;
/**
* Optional native atomic single-row claim. When omitted, the adapter
* Optional native atomic single-row consume. When omitted, the adapter
* factory falls back to `transaction(findMany + deleteMany)`.
* Implementing this method natively (e.g. `DELETE ... RETURNING *`,
* `findOneAndDelete`, `OUTPUT deleted.*`) gives one round trip and the
* strongest race-safety guarantee. Implementations must delete at most
* one matching row. TODO(claim-one-required): tighten to required in the
* one matching row. TODO(consume-one-required): tighten to required in the
* next minor on `next`.
*/
claimOne?: <T>(data: {
consumeOne?: <T>(data: {
model: string;
where: CleanedWhere[];
}) => Promise<T | null>;
+1 -1
View File
@@ -122,7 +122,7 @@ export type AdapterFactoryCustomizeAdapterCreator = (config: {
| "updateMany"
| "delete"
| "deleteMany"
| "claimOne"
| "consumeOne"
| "count";
}) => W extends undefined ? undefined : CleanedWhere[];
}) => CustomAdapter;
@@ -836,7 +836,7 @@ export const drizzleAdapter = (db: DB, config: DrizzleAdapterConfig) => {
}
return count;
},
async claimOne({ model, where }) {
async consumeOne({ model, where }) {
const schemaModel = getSchema(model);
const clause = convertWhereClause(where, model);
const idField = getFieldName({ model, field: "id" });
@@ -22,7 +22,7 @@ describe("kysely-adapter", () => {
expect(adapter).toBeDefined();
});
it("claimOne deletes only the selected row for non-unique predicates", async () => {
it("consumeOne deletes only the selected row for non-unique predicates", async () => {
const selectQuery = {
select: vi.fn(() => selectQuery),
where: vi.fn(() => selectQuery),
@@ -44,7 +44,7 @@ describe("kysely-adapter", () => {
} as any;
const adapter = kyselyAdapter(db)({});
const result = await adapter.claimOne({
const result = await adapter.consumeOne({
model: "verification",
where: [{ field: "identifier", value: "same-identifier" }],
});
@@ -663,7 +663,7 @@ export const kyselyAdapter = (
? Number.MAX_SAFE_INTEGER
: Number(res);
},
async claimOne({ model, where }) {
async consumeOne({ model, where }) {
const { and, or } = convertWhereClause(model, where);
const applyWhere = (query: any) => {
if (and) {
@@ -386,7 +386,7 @@ export const memoryAdapter = (
});
return count;
},
claimOne: async ({ model, where }) => {
consumeOne: async ({ model, where }) => {
const table = db[model]!;
const matches = convertWhereClause(where, model);
const target = matches[0];
@@ -11,7 +11,7 @@ describe("mongodb-adapter", () => {
expect(adapter).toBeDefined();
});
it("claimOne returns the deleted document from Mongo metadata", async () => {
it("consumeOne returns the deleted document from Mongo metadata", async () => {
const deleted = {
_id: "verification-id",
identifier: "magic-link-token",
@@ -24,7 +24,7 @@ describe("mongodb-adapter", () => {
} as any;
const adapter = mongodbAdapter(db, { transaction: false })({});
const result = await adapter.claimOne({
const result = await adapter.consumeOne({
model: "verification",
where: [{ field: "identifier", value: "magic-link-token" }],
});
@@ -630,7 +630,7 @@ export const mongodbAdapter = (
.deleteMany(clause, { session });
return res.deletedCount;
},
async claimOne({ model, where }) {
async consumeOne({ model, where }) {
const clause = convertWhereClause({ where, model });
const doc = await db.collection(model).findOneAndDelete(clause, {
session,
@@ -214,7 +214,7 @@ describe("prisma-adapter", () => {
});
});
it("claimOne rechecks non-unique predicates before deleting", async () => {
it("consumeOne rechecks non-unique predicates before deleting", async () => {
const target = {
id: "verification-id",
identifier: "magic-link-token",
@@ -234,7 +234,7 @@ describe("prisma-adapter", () => {
},
});
const result = await adapter.claimOne({
const result = await adapter.consumeOne({
model: "verification",
where: [{ field: "identifier", value: "magic-link-token" }],
});
@@ -251,7 +251,7 @@ describe("prisma-adapter", () => {
});
});
it("claimOne does not open a nested transaction from a transaction adapter", async () => {
it("consumeOne does not open a nested transaction from a transaction adapter", async () => {
const target = {
id: "verification-id",
identifier: "magic-link-token",
@@ -274,7 +274,7 @@ describe("prisma-adapter", () => {
)({} as BetterAuthOptions);
await adapter.transaction(async (trx) => {
await trx.claimOne({
await trx.consumeOne({
model: "verification",
where: [{ field: "identifier", value: "magic-link-token" }],
});
@@ -618,7 +618,7 @@ export const prismaAdapter = (prisma: PrismaClient, config: PrismaConfig) => {
});
return result ? (result.count as number) : 0;
},
async claimOne({ model, where }) {
async consumeOne({ model, where }) {
if (!db[model]) {
throw new BetterAuthError(
`Model ${model} does not exist in the database. If you haven't generated the Prisma client, you need to run 'npx prisma generate'`,
@@ -630,7 +630,7 @@ export const prismaAdapter = (prisma: PrismaClient, config: PrismaConfig) => {
// otherwise we fall back to find-then-deleteMany inside a
// transaction. deleteMany rechecks the original predicate with the
// selected id so we do not delete a row that stopped matching.
// FIXME(claim-one-prisma-locking): Prisma has no portable
// FIXME(consume-one-prisma-locking): Prisma has no portable
// row-locking API for findFirst. Add provider-specific locking
// when a breaking adapter contract can expose it cleanly.
const hasIdField = where?.some((w) => w.field === "id");