fix(drizzle-adapter): re-check the incrementOne guard on the UPDATE (#11331)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Gustavo Valverde <g.valverde02@gmail.com>
This commit is contained in:
Chris
2026-09-30 19:10:44 +00:00
committed by GitHub
co-authored by Claude Fable 5.1 Gustavo Valverde
parent 4186e36d15
commit 35d7cd3892
6 changed files with 620 additions and 16 deletions
+5
View File
@@ -0,0 +1,5 @@
---
"@better-auth/drizzle-adapter": patch
---
Drizzle `incrementOne` now rejects updates when a concurrent write makes the original `where` condition false. This prevents stale updates and counter limits from being exceeded on PostgreSQL. Both the default adapter and `relations-v2` retain the single-row limit.
@@ -0,0 +1,280 @@
/**
* @see https://github.com/better-auth/better-auth/issues/10557
*
* `incrementOne` is the adapter's compare-and-swap primitive: the `where`
* clause is the guard, and a call must fail if a concurrent writer makes
* that condition false. The rate limiter increments `count` guarded by `count < max`,
* the organization plugin flips an invitation guarded by `status = pending`,
* and two-factor consumes a backup code guarded by the old code list.
*
* The adapter selected one id under the guard in a subquery and updated by
* that id alone. Under READ COMMITTED, PostgreSQL re-checks an UPDATE's own
* WHERE against the new row version after waiting on a concurrent writer,
* but not an uncorrelated subquery's, so every waiting call succeeded once
* the first one committed: a limit of 5 let 8 of 20 through, and two callers
* both "claimed" one invitation. The guard now lives on the UPDATE as well.
*
* The same test exists for `relations-v2`, which has its own copy of the code.
*/
import { drizzleAdapter } from "@better-auth/drizzle-adapter";
import { drizzle } from "drizzle-orm/node-postgres";
import { integer, pgSchema, text } from "drizzle-orm/pg-core";
import type { PoolClient } from "pg";
import { Pool } from "pg";
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
// A schema of its own: the sibling suite in this directory runs
// `DROP SCHEMA public CASCADE` while files run in parallel.
const schema = pgSchema("increment_one_contention");
const counters = schema.table("counters", {
id: text("id").primaryKey(),
used: integer("used").notNull(),
status: text("status").notNull(),
});
// Every call must be in flight at once for the statements to contend on the
// row rather than queue in the pool; with a pool of 1 the unguarded adapter
// passes this test too.
const CONCURRENT_CALLS = 20;
const MAX = 5;
type CounterRow = { id: string; used: number; status: string };
describe("drizzle adapter (pg): incrementOne under contention", () => {
const pool = new Pool({
connectionString:
process.env.BETTER_AUTH_ADAPTER_TEST_DATABASE_URL ??
"postgres://user:password@localhost:5432/better_auth",
max: CONCURRENT_CALLS,
});
// The server can drop an idle client between queries; without a handler
// node-postgres emits an unhandled 'error' that ends the whole run.
pool.on("error", (error) => {
console.error("idle client error", error);
});
const createAdapter = (client: Pool | PoolClient) =>
drizzleAdapter(drizzle(client, { schema: { counters } }), {
provider: "pg",
schema: { counters },
})({
plugins: [
{
id: "counters-test",
schema: {
counters: {
fields: {
used: { type: "number" },
status: { type: "string" },
},
},
},
},
],
});
const adapter = createAdapter(pool);
const seed = (ids: string[]) =>
pool.query(
`INSERT INTO "increment_one_contention"."counters" ("id", "used", "status")
VALUES ${ids.map((_, i) => `($${i + 1}, 0, 'pending')`).join(", ")}`,
ids,
);
const rows = async () =>
(
await pool.query<{ id: string; used: number; status: string }>(
`SELECT "id", "used", "status" FROM "increment_one_contention"."counters" ORDER BY "id"`,
)
).rows;
const bump = (id: string) =>
adapter.incrementOne<{ used: number }>({
model: "counters",
where: [
{ field: "id", value: id },
{ field: "used", operator: "lt", value: MAX },
],
increment: { used: 1 },
});
async function whileGuardChanges(
operation: (
adapter: ReturnType<typeof createAdapter>,
) => Promise<CounterRow | null>,
) {
const blocker = await pool.connect();
const updater = await pool.connect();
let transactionOpen = false;
let pending: Promise<CounterRow | null> | undefined;
try {
await blocker.query("BEGIN");
transactionOpen = true;
await blocker.query(
`UPDATE "increment_one_contention"."counters" SET "used" = 0, "status" = 'accepted' WHERE "id" = 'shared'`,
);
const { rows } = await updater.query<{ pid: number }>(
"SELECT pg_backend_pid() AS pid",
);
pending = operation(createAdapter(updater));
void pending.catch(() => {});
// Keep the changed row uncommitted until the guarded statement is
// known to be waiting with the old condition in its snapshot.
const deadline = Date.now() + 3000;
let waiting = false;
while (Date.now() < deadline) {
const result = await pool.query<{ blocked: boolean }>(
"SELECT wait_event_type = 'Lock' AS blocked FROM pg_stat_activity WHERE pid = $1",
[rows[0]!.pid],
);
if (result.rows[0]?.blocked) {
waiting = true;
break;
}
await new Promise((resolve) => setTimeout(resolve, 10));
}
if (!waiting) {
throw new Error("The guarded update did not reach the held row lock");
}
await blocker.query("COMMIT");
transactionOpen = false;
return await pending;
} finally {
try {
if (transactionOpen) await blocker.query("ROLLBACK");
} finally {
await pending?.catch(() => {});
blocker.release();
updater.release();
}
}
}
beforeAll(async () => {
expect(pool.options.max).toBe(CONCURRENT_CALLS);
await pool.query(`
CREATE SCHEMA IF NOT EXISTS "increment_one_contention";
DROP TABLE IF EXISTS "increment_one_contention"."counters";
CREATE TABLE "increment_one_contention"."counters" (
"id" text PRIMARY KEY,
"used" integer NOT NULL,
"status" text NOT NULL
);
`);
});
beforeEach(async () => {
// Seeded directly: the adapter factory assigns its own ids on `create`,
// and the guards below address rows by these.
await pool.query(`DELETE FROM "increment_one_contention"."counters"`);
});
afterAll(async () => {
await pool.query(
`DROP SCHEMA IF EXISTS "increment_one_contention" CASCADE`,
);
await pool.end();
});
it("lets a bounded counter be incremented exactly `max` times", async () => {
await seed(["shared"]);
const results = await Promise.all(
Array.from({ length: CONCURRENT_CALLS }, () => bump("shared")),
);
expect(results.filter((result) => result !== null)).toHaveLength(MAX);
expect(await rows()).toEqual([
{ id: "shared", used: MAX, status: "pending" },
]);
});
it("mutates one row per win when the guard matches several rows", async () => {
// A guard with no id in it matches every row that is still under the
// limit. The `LIMIT 1` subquery pins each call to one row: every win must
// move exactly one row, and no row may pass the limit.
await seed(["a", "b", "c"]);
const results = await Promise.all(
Array.from({ length: CONCURRENT_CALLS }, () =>
adapter.incrementOne<{ used: number }>({
model: "counters",
where: [{ field: "used", operator: "lt", value: MAX }],
increment: { used: 1 },
}),
),
);
const after = await rows();
const wins = results.filter((result) => result !== null).length;
expect(after.reduce((sum, row) => sum + row.used, 0)).toBe(wins);
for (const row of after) expect(row.used).toBeLessThanOrEqual(MAX);
});
it("lets exactly one caller win a status transition", async () => {
await seed(["shared"]);
const results = await Promise.all(
Array.from({ length: CONCURRENT_CALLS }, () =>
adapter.incrementOne<{ status: string }>({
model: "counters",
where: [
{ field: "id", value: "shared" },
{ field: "status", value: "pending" },
],
increment: {},
set: { status: "accepted" },
}),
),
);
expect(results.filter((result) => result !== null)).toHaveLength(1);
expect(await rows()).toEqual([
{ id: "shared", used: 0, status: "accepted" },
]);
});
it("rejects a set-only update whose guard changes while waiting", async () => {
await seed(["shared"]);
const result = await whileGuardChanges((adapter) =>
adapter.incrementOne<CounterRow>({
model: "counters",
where: [
{ field: "id", value: "shared" },
{ field: "status", value: "pending" },
],
increment: {},
set: { status: "stale" },
}),
);
expect(result).toBeNull();
expect(await rows()).toEqual([
{ id: "shared", used: 0, status: "accepted" },
]);
});
it("rejects a decrement whose guard changes while waiting", async () => {
await seed(["shared"]);
await pool.query(
`UPDATE "increment_one_contention"."counters" SET "used" = 1 WHERE "id" = 'shared'`,
);
const result = await whileGuardChanges((adapter) =>
adapter.incrementOne<CounterRow>({
model: "counters",
where: [
{ field: "id", value: "shared" },
{ field: "used", operator: "gt", value: 0 },
],
increment: { used: -1 },
}),
);
expect(result).toBeNull();
expect(await rows()).toEqual([
{ id: "shared", used: 0, status: "accepted" },
]);
});
});
@@ -0,0 +1,281 @@
/**
* @see https://github.com/better-auth/better-auth/issues/10557
*
* `incrementOne` is the adapter's compare-and-swap primitive: the `where`
* clause is the guard, and a call must fail if a concurrent writer makes
* that condition false. The rate limiter increments `count` guarded by `count < max`,
* the organization plugin flips an invitation guarded by `status = pending`,
* and two-factor consumes a backup code guarded by the old code list.
*
* The adapter selected one id under the guard in a subquery and updated by
* that id alone. Under READ COMMITTED, PostgreSQL re-checks an UPDATE's own
* WHERE against the new row version after waiting on a concurrent writer,
* but not an uncorrelated subquery's, so every waiting call succeeded once
* the first one committed: a limit of 5 let 8 of 20 through, and two callers
* both "claimed" one invitation. The guard now lives on the UPDATE as well.
*
* `relations-v2` carries its own copy of `incrementOne`; this is the same test
* as the one in ../drizzle-adapter, against that entry point and drizzle 1.0.
*/
import { drizzleAdapter } from "@better-auth/drizzle-adapter/relations-v2";
import { drizzle } from "drizzle-orm/node-postgres";
import { integer, pgSchema, text } from "drizzle-orm/pg-core";
import type { PoolClient } from "pg";
import { Pool } from "pg";
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
// A schema of its own: the sibling suite in this directory runs
// `DROP SCHEMA public CASCADE` while files run in parallel.
const schema = pgSchema("increment_one_contention_v2");
const counters = schema.table("counters", {
id: text("id").primaryKey(),
used: integer("used").notNull(),
status: text("status").notNull(),
});
// Every call must be in flight at once for the statements to contend on the
// row rather than queue in the pool; with a pool of 1 the unguarded adapter
// passes this test too.
const CONCURRENT_CALLS = 20;
const MAX = 5;
type CounterRow = { id: string; used: number; status: string };
describe("drizzle adapter relations-v2 (pg): incrementOne under contention", () => {
const pool = new Pool({
connectionString:
process.env.BETTER_AUTH_ADAPTER_TEST_DATABASE_URL ??
"postgres://user:password@localhost:5432/better_auth",
max: CONCURRENT_CALLS,
});
// The server can drop an idle client between queries; without a handler
// node-postgres emits an unhandled 'error' that ends the whole run.
pool.on("error", (error) => {
console.error("idle client error", error);
});
const createAdapter = (client: Pool | PoolClient) =>
drizzleAdapter(drizzle({ client }), {
provider: "pg",
schema: { counters },
})({
plugins: [
{
id: "counters-test",
schema: {
counters: {
fields: {
used: { type: "number" },
status: { type: "string" },
},
},
},
},
],
});
const adapter = createAdapter(pool);
const seed = (ids: string[]) =>
pool.query(
`INSERT INTO "increment_one_contention_v2"."counters" ("id", "used", "status")
VALUES ${ids.map((_, i) => `($${i + 1}, 0, 'pending')`).join(", ")}`,
ids,
);
const rows = async () =>
(
await pool.query<{ id: string; used: number; status: string }>(
`SELECT "id", "used", "status" FROM "increment_one_contention_v2"."counters" ORDER BY "id"`,
)
).rows;
const bump = (id: string) =>
adapter.incrementOne<{ used: number }>({
model: "counters",
where: [
{ field: "id", value: id },
{ field: "used", operator: "lt", value: MAX },
],
increment: { used: 1 },
});
async function whileGuardChanges(
operation: (
adapter: ReturnType<typeof createAdapter>,
) => Promise<CounterRow | null>,
) {
const blocker = await pool.connect();
const updater = await pool.connect();
let transactionOpen = false;
let pending: Promise<CounterRow | null> | undefined;
try {
await blocker.query("BEGIN");
transactionOpen = true;
await blocker.query(
`UPDATE "increment_one_contention_v2"."counters" SET "used" = 0, "status" = 'accepted' WHERE "id" = 'shared'`,
);
const { rows } = await updater.query<{ pid: number }>(
"SELECT pg_backend_pid() AS pid",
);
pending = operation(createAdapter(updater));
void pending.catch(() => {});
// Keep the changed row uncommitted until the guarded statement is
// known to be waiting with the old condition in its snapshot.
const deadline = Date.now() + 3000;
let waiting = false;
while (Date.now() < deadline) {
const result = await pool.query<{ blocked: boolean }>(
"SELECT wait_event_type = 'Lock' AS blocked FROM pg_stat_activity WHERE pid = $1",
[rows[0]!.pid],
);
if (result.rows[0]?.blocked) {
waiting = true;
break;
}
await new Promise((resolve) => setTimeout(resolve, 10));
}
if (!waiting) {
throw new Error("The guarded update did not reach the held row lock");
}
await blocker.query("COMMIT");
transactionOpen = false;
return await pending;
} finally {
try {
if (transactionOpen) await blocker.query("ROLLBACK");
} finally {
await pending?.catch(() => {});
blocker.release();
updater.release();
}
}
}
beforeAll(async () => {
expect(pool.options.max).toBe(CONCURRENT_CALLS);
await pool.query(`
CREATE SCHEMA IF NOT EXISTS "increment_one_contention_v2";
DROP TABLE IF EXISTS "increment_one_contention_v2"."counters";
CREATE TABLE "increment_one_contention_v2"."counters" (
"id" text PRIMARY KEY,
"used" integer NOT NULL,
"status" text NOT NULL
);
`);
});
beforeEach(async () => {
// Seeded directly: the adapter factory assigns its own ids on `create`,
// and the guards below address rows by these.
await pool.query(`DELETE FROM "increment_one_contention_v2"."counters"`);
});
afterAll(async () => {
await pool.query(
`DROP SCHEMA IF EXISTS "increment_one_contention_v2" CASCADE`,
);
await pool.end();
});
it("lets a bounded counter be incremented exactly `max` times", async () => {
await seed(["shared"]);
const results = await Promise.all(
Array.from({ length: CONCURRENT_CALLS }, () => bump("shared")),
);
expect(results.filter((result) => result !== null)).toHaveLength(MAX);
expect(await rows()).toEqual([
{ id: "shared", used: MAX, status: "pending" },
]);
});
it("mutates one row per win when the guard matches several rows", async () => {
// A guard with no id in it matches every row that is still under the
// limit. The `LIMIT 1` subquery pins each call to one row: every win must
// move exactly one row, and no row may pass the limit.
await seed(["a", "b", "c"]);
const results = await Promise.all(
Array.from({ length: CONCURRENT_CALLS }, () =>
adapter.incrementOne<{ used: number }>({
model: "counters",
where: [{ field: "used", operator: "lt", value: MAX }],
increment: { used: 1 },
}),
),
);
const after = await rows();
const wins = results.filter((result) => result !== null).length;
expect(after.reduce((sum, row) => sum + row.used, 0)).toBe(wins);
for (const row of after) expect(row.used).toBeLessThanOrEqual(MAX);
});
it("lets exactly one caller win a status transition", async () => {
await seed(["shared"]);
const results = await Promise.all(
Array.from({ length: CONCURRENT_CALLS }, () =>
adapter.incrementOne<{ status: string }>({
model: "counters",
where: [
{ field: "id", value: "shared" },
{ field: "status", value: "pending" },
],
increment: {},
set: { status: "accepted" },
}),
),
);
expect(results.filter((result) => result !== null)).toHaveLength(1);
expect(await rows()).toEqual([
{ id: "shared", used: 0, status: "accepted" },
]);
});
it("rejects a set-only update whose guard changes while waiting", async () => {
await seed(["shared"]);
const result = await whileGuardChanges((adapter) =>
adapter.incrementOne<CounterRow>({
model: "counters",
where: [
{ field: "id", value: "shared" },
{ field: "status", value: "pending" },
],
increment: {},
set: { status: "stale" },
}),
);
expect(result).toBeNull();
expect(await rows()).toEqual([
{ id: "shared", used: 0, status: "accepted" },
]);
});
it("rejects a decrement whose guard changes while waiting", async () => {
await seed(["shared"]);
await pool.query(
`UPDATE "increment_one_contention_v2"."counters" SET "used" = 1 WHERE "id" = 'shared'`,
);
const result = await whileGuardChanges((adapter) =>
adapter.incrementOne<CounterRow>({
model: "counters",
where: [
{ field: "id", value: "shared" },
{ field: "used", operator: "gt", value: 0 },
],
increment: { used: -1 },
}),
);
expect(result).toBeNull();
expect(await rows()).toEqual([
{ id: "shared", used: 0, status: "accepted" },
]);
});
});
@@ -529,7 +529,13 @@ describe("drizzle-adapter", () => {
});
});
describe("incrementOne", () => {
// Both entry points carry their own copy of incrementOne, so every case
// below runs against each; a fix landing in one copy but not the other is
// caught here, without a database.
describe.each([
{ relations: "Relations v1", adapterFactory: drizzleAdapter },
{ relations: "Relations v2", adapterFactory: drizzleRelationsV2Adapter },
])("incrementOne ($relations)", ({ adapterFactory }) => {
const defaultSecret = "test-secret-that-is-at-least-32-chars-long!!";
const userTable = pgTable("user", {
id: text("id"),
@@ -545,10 +551,11 @@ describe("drizzle-adapter", () => {
/**
* Builds a mock db that mirrors the adapter's single-row update: a
* `select().from().where().limit()` subquery picks one id, then
* `update().set().where().returning()` mutates by that id. Captures the
* `set` payload, the update's `where` args, and the select guard so a test
* can assert the `field = field + delta` expression and that the update is
* pinned to one selected id rather than the raw guard clause.
* `update().set().where().returning()` mutates under the guard AND that
* id. Captures the `set` payload, the update's `where` args, and the
* select guard so a test can assert the `field = field + delta`
* expression and that the update repeats the guard alongside the pinned
* id, rather than trusting the subquery alone.
*/
function createIncrementDb(returned: unknown[]) {
const calls: {
@@ -580,8 +587,15 @@ describe("drizzle-adapter", () => {
return { db, calls, targetIds };
}
/** Every chunk of an SQL expression, descending into nested SQL. */
function flattenSqlChunks(expr: SQL): unknown[] {
return expr.queryChunks.flatMap((chunk) =>
is(chunk, SQL) ? [chunk, ...flattenSqlChunks(chunk)] : [chunk],
);
}
function createAdapter(db: any) {
return drizzleAdapter(db, { provider: "sqlite" })({
return adapterFactory(db, { provider: "sqlite" })({
secret: defaultSecret,
user: {
additionalFields: {
@@ -616,8 +630,8 @@ describe("drizzle-adapter", () => {
expect(
chunks.some((chunk) => is(chunk, Param) && chunk.value === 3),
).toBe(true);
// The guard runs on the SELECT that picks one id (one predicate here);
// the UPDATE is pinned to that single id, not the raw guard clause.
// The guard runs on the SELECT that picks one id (one predicate here),
// and again on the UPDATE together with that id (one combined predicate).
expect(calls.selectGuard).toHaveLength(1);
expect(calls.whereArgs).toHaveLength(1);
});
@@ -658,18 +672,28 @@ describe("drizzle-adapter", () => {
expect(result).toEqual({ id: "user-1", attempts: 5 });
// The non-unique guard is applied to the SELECT, which is capped to one
// row; the UPDATE never receives the raw guard.
// row, so the UPDATE can touch at most that row.
expect(db.select).toHaveBeenCalledTimes(1);
expect(calls.selectGuard).toHaveLength(1);
// The UPDATE is guarded by a single `id IN (<one-row subquery>)`
// predicate, not the original multi-row clause.
// The UPDATE carries one combined predicate: the pinned single-id
// subquery AND the original guard. Both are required. The subquery keeps
// the mutation to one row. Repeating the guard is what makes the call a
// compare-and-swap under concurrency: on PostgreSQL, an UPDATE that
// waited for a concurrent writer re-checks its own WHERE against the new
// row version, but not an uncorrelated subquery's — so a guard that lives
// only in the subquery is never re-evaluated, and every waiting increment
// succeeds after the first one commits (#10557).
expect(calls.whereArgs).toHaveLength(1);
const updateGuard = calls.whereArgs?.[0];
expect(is(updateGuard, SQL)).toBe(true);
// The pinned predicate embeds the single-id subquery, proving the update
// targets only the one selected row.
expect((updateGuard as SQL).queryChunks).toContain(targetIds);
const predicate = flattenSqlChunks(updateGuard as SQL);
expect(predicate).toContain(targetIds);
expect(predicate).toContain(calls.selectGuard?.[0]);
// ...and they are conjoined: `guard OR id IN (...)` would match by id
// alone and reopen the race.
expect(predicate).toContainEqual({ value: [" and "] });
expect(predicate).not.toContainEqual({ value: [" or "] });
});
it("returns null when the guard matches no row", async () => {
@@ -1204,10 +1204,19 @@ export const drizzleAdapter = (db: DB, config: DrizzleAdapterConfig) => {
.from(table)
.where(...clause)
.limit(1);
// The guard is repeated on the UPDATE itself, as the kysely adapter
// does on this path. Under READ COMMITTED, PostgreSQL re-evaluates an
// UPDATE's own WHERE against the new row version after waiting on a
// concurrent writer, but not an uncorrelated subquery's: with the
// guard only in `targetIds`, `id IN (...)` still matched and every
// waiting compare-and-swap succeeded once the first one committed
// (#10557). The MySQL branch above needs no such re-check: it holds
// the row under SELECT ... FOR UPDATE in the same transaction.
// ./relations-v2/index.ts carries its own copy of this method.
const updated = await db
.update(table)
.set(assignments)
.where(inArray(idColumn, targetIds))
.where(and(...clause, inArray(idColumn, targetIds)))
.returning();
return (updated[0] as any) ?? null;
},
@@ -1129,10 +1129,15 @@ export const drizzleAdapter = (db: DB, config: DrizzleAdapterConfig) => {
.from(table)
.where(...clause)
.limit(1);
// The guard is repeated on the UPDATE itself: PostgreSQL re-checks
// an UPDATE's own WHERE after waiting on a concurrent writer, but not
// an uncorrelated subquery's, so a guard that lives only in
// `targetIds` is never re-evaluated (#10557). See the same code in
// ../drizzle-adapter.ts.
const updated = await db
.update(table)
.set(assignments)
.where(inArray(idColumn, targetIds))
.where(and(...clause, inArray(idColumn, targetIds)))
.returning();
return (updated[0] as any) ?? null;
},