diff --git a/.changeset/one-counter-every-instance-reads.md b/.changeset/one-counter-every-instance-reads.md new file mode 100644 index 0000000000..ba20930f39 --- /dev/null +++ b/.changeset/one-counter-every-instance-reads.md @@ -0,0 +1,60 @@ +--- +"@nextlyhq/adapter-drizzle": patch +"@nextlyhq/adapter-mysql": patch +"@nextlyhq/adapter-postgres": patch +"@nextlyhq/adapter-sqlite": patch +"@nextlyhq/admin": patch +"@nextlyhq/admin-css": patch +"@nextlyhq/blocks-engine": patch +"@nextlyhq/blocks-react": patch +"@nextlyhq/builder": patch +"@nextlyhq/eslint-config": patch +"@nextlyhq/eslint-plugin": patch +"@nextlyhq/module-specifiers": patch +"@nextlyhq/plugin-form-builder": patch +"@nextlyhq/plugin-mcp": patch +"@nextlyhq/plugin-page-builder": patch +"@nextlyhq/plugin-sdk": patch +"@nextlyhq/plugin-seo": patch +"@nextlyhq/prettier-config": patch +"@nextlyhq/storage-s3": patch +"@nextlyhq/storage-uploadthing": patch +"@nextlyhq/storage-vercel-blob": patch +"@nextlyhq/telemetry": patch +"@nextlyhq/tsconfig": patch +"@nextlyhq/ui": patch +"create-nextly-app": patch +"nextly": patch +--- + +The signal that retires a cached authorization answer is now stored in the +database, so every instance sees it. It was a counter held in memory, which +moved only in the process that handled the change: a second instance neither +saw the move nor had one of its own, and went on serving what it had cached +until the entry aged out. On the shared tier that meant a revoked grant could +outlive its revocation by the whole cache lifetime. + +Cross-instance revocation now takes effect within about a second. Each instance +reads the shared counter at most once per second rather than once per check, so +the cost is one small indexed read per second per instance and not one per +request. The instance that MADE the change applies it immediately. + +Two behaviour changes worth knowing about. + +An invalidation naming one user now retires every in-memory answer rather than +that user's alone. The counter other instances read carries a number and not a +user id, so a change they can see cannot be narrower than "something in RBAC +moved", and keeping the scope locally would mean only the instance that made the +change applied it narrowly. Refilling is a couple of indexed queries and role +changes are rare; a stale answer costs a grant the install revoked. + +A batch of permission writes no longer holds back the in-memory tiers. It never +existed to: what it saves is the unfiltered rewrite of every stored row, and +that is still deferred to the end of the batch. + +Installations upgraded from an earlier version keep working before they +reconcile their core tables. The new table arrives through `nextly db:sync`, and +until it does, every read and write of the counter degrades to the previous +in-memory behaviour rather than failing the authorization check that asked. The +degraded state is reported once so an operator can see why cross-instance +invalidation is not yet in effect. diff --git a/packages/nextly/src/database/sqlite-core-tables.ts b/packages/nextly/src/database/sqlite-core-tables.ts index 4d68f63ca0..259668be89 100644 --- a/packages/nextly/src/database/sqlite-core-tables.ts +++ b/packages/nextly/src/database/sqlite-core-tables.ts @@ -1,3 +1,4 @@ +import { RBAC_EPOCH_TABLE } from "../schemas/rbac-epoch/table-name"; import { STORAGE_FORMAT } from "../schemas/storage-format"; // Raw CREATE TABLE IF NOT EXISTS DDL for all Nextly core SQLite tables. @@ -238,6 +239,22 @@ export function generateSqliteCoreTableStatements(): string[] { ON "user_permission_cache" ("expires_at")`, `CREATE INDEX IF NOT EXISTS "upc_user_action_resource_idx" ON "user_permission_cache" ("user_id", "action", "resource")`, + // The counter every instance reads to decide whether a cached + // authorization answer is still current. One row, keyed `global`: the + // primary key is what makes a second counter unrepresentable, so nothing + // has to keep "there is exactly one" true. + // + // Reached by an existing installation because this whole bootstrap is + // re-run as a reconciliation, and `IF NOT EXISTS` adds only what is absent. + // That is why the epoch is a TABLE rather than a column on the rows above: + // SQLite skips a `CREATE TABLE` wholesale once the table exists, so a new + // column there would never arrive. + `CREATE TABLE IF NOT EXISTS "${RBAC_EPOCH_TABLE}" ( + "id" TEXT PRIMARY KEY, + "revision" INTEGER NOT NULL, + "generation" TEXT NOT NULL, + "updated_at" INTEGER NOT NULL + )`, `CREATE TABLE IF NOT EXISTS "content_schema_events" ( "id" INTEGER PRIMARY KEY AUTOINCREMENT, "op" TEXT NOT NULL, diff --git a/packages/nextly/src/domains/auth/services/api-key-service.ts b/packages/nextly/src/domains/auth/services/api-key-service.ts index 17c631c36c..9f2c38e139 100644 --- a/packages/nextly/src/domains/auth/services/api-key-service.ts +++ b/packages/nextly/src/domains/auth/services/api-key-service.ts @@ -69,8 +69,8 @@ import { BaseService } from "../../../services/base-service"; import { isSuperAdmin, listRoleSlugsForUserOrRefuse, - rbacRevision, } from "../../../services/lib/permissions"; +import { refreshEpoch, stampIsCurrent } from "../../../services/lib/rbac-epoch"; import type { Logger } from "../../../services/shared"; /** The three token types that determine how permissions are resolved at request time. */ @@ -281,7 +281,7 @@ const _apiKeyPermissionsCache = new Map< * old set, which the comment in `UserRoleService` said was handled * elsewhere and was not. */ - revision: number; + revision: string; } >(); const _PERMISSIONS_CACHE_TTL_MS = 5 * 60 * 1000; @@ -748,20 +748,37 @@ export class ApiKeyService extends BaseService { keyId: string ): Promise { const cacheKey = `apikey:${keyId}`; - const now = Date.now(); // Read BEFORE the queries below, never after. An invalidation that lands // while they are in flight would otherwise be stamped onto the result they // return: the rows were read under the old revision and would be filed // under the new one, so the next request reuses grants the change was // meant to retire, for the whole TTL. Captured here, that entry is already // behind when it is written and the next read re-resolves. - const resolvedUnder = rbacRevision(); + // + // Refreshed rather than read, which is what makes the comparison below + // answer for the INSTALL rather than for this process. These grants are a + // copy of the catalogue held for five minutes, and a role revoked on + // another instance has to retire them here too. + const resolvedUnder = await refreshEpoch(); + + // Read AFTER the refresh, not before it. That refresh can wait — on a slow + // database, or behind another forced read it queued for — and an age taken + // beforehand is the age the entry had when the request started rather than + // when it is being served, so an entry that expired during the wait passes + // the window one more time. + const now = Date.now(); const cached = _apiKeyPermissionsCache.get(cacheKey); if ( cached && - now - cached.cachedAt < _PERMISSIONS_CACHE_TTL_MS && - cached.revision === resolvedUnder + // Asked through the shared predicate, not by comparing the stamp. A + // match means nothing while this process holds invalidations the shared + // row has not accepted: the value being matched against is then one no + // other instance has seen, so a revocation made here keeps answering + // from this copy for the whole window. The permission tiers ask the same + // question and this one was not. + stampIsCurrent(cached.revision) && + now - cached.cachedAt < _PERMISSIONS_CACHE_TTL_MS ) { return cached.grants; } diff --git a/packages/nextly/src/domains/auth/services/permission-cache-service.ts b/packages/nextly/src/domains/auth/services/permission-cache-service.ts index 97bce0c028..38e376b8ae 100644 --- a/packages/nextly/src/domains/auth/services/permission-cache-service.ts +++ b/packages/nextly/src/domains/auth/services/permission-cache-service.ts @@ -288,18 +288,16 @@ export class PermissionCacheService extends BaseService { * sees an expired row rather than a missing one. */ async invalidateAll(): Promise { - try { - const { userPermissionCache } = this.tables; - const result = await this.db - .update(userPermissionCache) - .set({ expiresAt: new Date() }); - return affectedRowCount(result, this.dialect); - } catch (error) { - this.logger.error("Failed to invalidate the whole permission cache", { - error: String(error), - }); - return 0; - } + // A failure is RAISED rather than reported as zero. Zero is also what a + // successful tombstone over an empty table answers, so swallowing the + // error makes the two indistinguishable — and the caller that publishes the + // epoch afterwards then announces a retirement that did not happen, which + // is the one announcement that must never be made. + const { userPermissionCache } = this.tables; + const result = await this.db + .update(userPermissionCache) + .set({ expiresAt: new Date() }); + return affectedRowCount(result, this.dialect); } async invalidateByUser(userId: string): Promise { @@ -307,36 +305,27 @@ export class PermissionCacheService extends BaseService { return 0; } - try { - const { userPermissionCache } = this.tables; + // Raised rather than reported as zero, for the reason `invalidateAll` + // gives: zero is also a successful tombstone that matched no rows. + const { userPermissionCache } = this.tables; - // Write-through invalidation: mark as expired (tombstone) instead of deleting - const result = await this.db - .update(userPermissionCache) - .set({ expiresAt: new Date() }) - .where(eq(userPermissionCache.userId, userId)); + // Write-through invalidation: mark as expired (tombstone) instead of deleting + const result = await this.db + .update(userPermissionCache) + .set({ expiresAt: new Date() }) + .where(eq(userPermissionCache.userId, userId)); - const invalidatedCount = affectedRowCount(result, this.dialect); - - if (process.env.DEBUG_CACHE === "1") { - console.log("[cache][dbg] invalidateByUser", { - userId, - invalidatedCount, - method: "tombstone", - }); - } + const invalidatedCount = affectedRowCount(result, this.dialect); - return invalidatedCount; - } catch (error) { - getAuthLogger()?.log?.("error", { - category: "auth", - op: "cache", - message: "invalidateByUser failed", + if (process.env.DEBUG_CACHE === "1") { + console.log("[cache][dbg] invalidateByUser", { userId, - error: String(error), + invalidatedCount, + method: "tombstone", }); - return 0; } + + return invalidatedCount; } /** @@ -364,7 +353,9 @@ export class PermissionCacheService extends BaseService { return 0; } - try { + // Raised rather than reported as zero, for the reason `invalidateAll` + // gives: zero is also a successful tombstone that matched no rows. + { const { userPermissionCache } = this.tables; // Write-through invalidation: mark as expired (tombstone) instead of deleting. @@ -408,15 +399,6 @@ export class PermissionCacheService extends BaseService { } return invalidatedCount; - } catch (error) { - getAuthLogger()?.log?.("error", { - category: "auth", - op: "cache", - message: "invalidateByRole failed", - roleId, - error: String(error), - }); - return 0; } } diff --git a/packages/nextly/src/domains/auth/services/role-inheritance-service.ts b/packages/nextly/src/domains/auth/services/role-inheritance-service.ts index a48a74b55f..77d3fcc3b7 100644 --- a/packages/nextly/src/domains/auth/services/role-inheritance-service.ts +++ b/packages/nextly/src/domains/auth/services/role-inheritance-service.ts @@ -96,7 +96,7 @@ export class RoleInheritanceService extends BaseService { } } - void invalidatePermissionCache({ roleId: childRoleId }); + await invalidatePermissionCache({ roleId: childRoleId }); } /** @@ -118,7 +118,7 @@ export class RoleInheritanceService extends BaseService { ) ); - void invalidatePermissionCache({ roleId: childRoleId }); + await invalidatePermissionCache({ roleId: childRoleId }); } /** diff --git a/packages/nextly/src/domains/auth/services/role-permission-service.ts b/packages/nextly/src/domains/auth/services/role-permission-service.ts index afa317a407..1aaa69c44b 100644 --- a/packages/nextly/src/domains/auth/services/role-permission-service.ts +++ b/packages/nextly/src/domains/auth/services/role-permission-service.ts @@ -160,7 +160,7 @@ export class RolePermissionService extends BaseService { permissionId = newPermId; } - void invalidatePermissionCache({ roleId }); + await invalidatePermissionCache({ roleId }); } /** @@ -244,7 +244,7 @@ export class RolePermissionService extends BaseService { ) ); - void invalidatePermissionCache({ roleId }); + await invalidatePermissionCache({ roleId }); } /** @@ -285,7 +285,7 @@ export class RolePermissionService extends BaseService { } } - void invalidatePermissionCache({ roleId }); + await invalidatePermissionCache({ roleId }); return this.listRolePermissions(roleId); } diff --git a/packages/nextly/src/domains/auth/services/role/role-mutation-service.ts b/packages/nextly/src/domains/auth/services/role/role-mutation-service.ts index fcfe1f1d13..a00d59d018 100644 --- a/packages/nextly/src/domains/auth/services/role/role-mutation-service.ts +++ b/packages/nextly/src/domains/auth/services/role/role-mutation-service.ts @@ -340,10 +340,11 @@ export class RoleMutationService extends BaseService { : // eslint-disable-next-line @typescript-eslint/no-explicit-any -- Drizzle transaction callback type varies by dialect await this.db.transaction(async (tx: any) => runMutations(tx)); - // Invalidate cache after successful transaction. `void` marks the - // promise as intentionally unawaited - cache invalidation is - // fire-and-forget and must not block the create response. - void invalidatePermissionCache({ roleId: id }); + // Awaited, and the ordering is the invariant rather than a preference: + // a runtime that freezes after responding can abandon an unawaited + // shared write, leaving every other instance on the old epoch with no + // sign anything went wrong. + await invalidatePermissionCache({ roleId: id }); return { id, @@ -605,7 +606,7 @@ export class RoleMutationService extends BaseService { changes.permissionIds !== undefined || changes.childRoleIds !== undefined ) { - void invalidatePermissionCache({ roleId }); + await invalidatePermissionCache({ roleId }); } return; @@ -689,8 +690,9 @@ export class RoleMutationService extends BaseService { await tx.delete("roles", this.whereEq("id", roleId)); }); - // Invalidate cache after successful transaction (fire-and-forget). - void invalidatePermissionCache({ roleId }); + // Awaited, for the reason `createRole` gives: an abandoned shared write + // leaves the other instances holding answers this delete retired. + await invalidatePermissionCache({ roleId }); } catch (e: unknown) { // Re-throw NextlyErrors unchanged. Raw DB errors map via // fromDatabaseError, which provides the spec-compliant generic public diff --git a/packages/nextly/src/domains/auth/services/user-role-service.ts b/packages/nextly/src/domains/auth/services/user-role-service.ts index 0f1df7a906..6f247a37f7 100644 --- a/packages/nextly/src/domains/auth/services/user-role-service.ts +++ b/packages/nextly/src/domains/auth/services/user-role-service.ts @@ -115,7 +115,7 @@ export class UserRoleService extends BaseService { await insert; } - void invalidatePermissionCache({ userId }); + await invalidatePermissionCache({ userId }); // Invalidate API key permission caches for this user's read-only and // full-access keys — their effective permissions derive from the creator's @@ -175,7 +175,7 @@ export class UserRoleService extends BaseService { ) ); - void invalidatePermissionCache({ userId }); + await invalidatePermissionCache({ userId }); // Invalidate API key permission caches for this user's read-only and // full-access keys — their effective permissions derive from the creator's diff --git a/packages/nextly/src/schemas/_dialect-bundles/mysql.ts b/packages/nextly/src/schemas/_dialect-bundles/mysql.ts index 48aaef3dfa..e8d00be8cd 100644 --- a/packages/nextly/src/schemas/_dialect-bundles/mysql.ts +++ b/packages/nextly/src/schemas/_dialect-bundles/mysql.ts @@ -17,6 +17,7 @@ export { users, accounts, sessions } from "../users/mysql"; // `reconcileCore` hands drizzle-kit. export { nextlyFieldGroupLock } from "../field-group-lock/mysql"; export { nextlyDocumentLock } from "../document-lock/mysql"; +export { nextlyRbacEpoch } from "../rbac-epoch/mysql"; export { nextlyWidgetLayout } from "../widget-layout/mysql"; export { diff --git a/packages/nextly/src/schemas/_dialect-bundles/postgres.ts b/packages/nextly/src/schemas/_dialect-bundles/postgres.ts index d703a81ce8..42da423231 100644 --- a/packages/nextly/src/schemas/_dialect-bundles/postgres.ts +++ b/packages/nextly/src/schemas/_dialect-bundles/postgres.ts @@ -32,6 +32,7 @@ export { users, accounts, sessions } from "../users/postgres"; // `reconcileCore` hands drizzle-kit. export { nextlyFieldGroupLock } from "../field-group-lock/postgres"; export { nextlyDocumentLock } from "../document-lock/postgres"; +export { nextlyRbacEpoch } from "../rbac-epoch/postgres"; export { nextlyWidgetLayout } from "../widget-layout/postgres"; // Auth tokens. diff --git a/packages/nextly/src/schemas/_dialect-bundles/sqlite.ts b/packages/nextly/src/schemas/_dialect-bundles/sqlite.ts index 2274d8f972..9c29d99ab6 100644 --- a/packages/nextly/src/schemas/_dialect-bundles/sqlite.ts +++ b/packages/nextly/src/schemas/_dialect-bundles/sqlite.ts @@ -17,6 +17,7 @@ export { users, accounts, sessions } from "../users/sqlite"; // `reconcileCore` hands drizzle-kit. export { nextlyFieldGroupLock } from "../field-group-lock/sqlite"; export { nextlyDocumentLock } from "../document-lock/sqlite"; +export { nextlyRbacEpoch } from "../rbac-epoch/sqlite"; export { nextlyWidgetLayout } from "../widget-layout/sqlite"; export { diff --git a/packages/nextly/src/schemas/index.ts b/packages/nextly/src/schemas/index.ts index d42eb45254..6e07c9a16b 100644 --- a/packages/nextly/src/schemas/index.ts +++ b/packages/nextly/src/schemas/index.ts @@ -56,6 +56,7 @@ import { mediaTables } from "./media"; import { nextlyI18nArchiveTables } from "./nextly-i18n-archive"; import { nextlyMetaTables } from "./nextly-meta"; import { rbacTables } from "./rbac"; +import { RBAC_EPOCH_TABLE, rbacEpochTables } from "./rbac-epoch"; import { releasesTables } from "./releases"; import { schemaEventsTables } from "./schema-events"; import { siteSettingsMysql } from "./site-settings/mysql"; @@ -175,6 +176,12 @@ export function getCoreSchema( // outside it is never created on a real installation, however completely // its own module declares it. ...Object.values(documentLockTables(dialect)), + // `nextly_rbac_epoch` — the one counter every instance reads to decide + // whether a cached authorization answer is still current. Declared here for + // the same reason as the lock above: a table outside this set is never + // created on a real installation, however completely its own module + // declares it. + ...Object.values(rbacEpochTables(dialect)), ...Object.values(apiKeyTables(dialect)), // `nextly_schema_events` (the migration ledger) is a first-class managed // table. It is still bootstrapped out-of-band via `getSchemaEventsDdl` so @@ -295,6 +302,7 @@ export const CORE_TABLE_NAMES: readonly string[] = [ // snapshot, so the drift check proposes adding it again on every run. "nextly_field_group_lock", "nextly_document_lock", + RBAC_EPOCH_TABLE, "nextly_widget_layout", "dynamic_collections", "dynamic_singles", diff --git a/packages/nextly/src/schemas/rbac-epoch/index.ts b/packages/nextly/src/schemas/rbac-epoch/index.ts new file mode 100644 index 0000000000..f7e14151c6 --- /dev/null +++ b/packages/nextly/src/schemas/rbac-epoch/index.ts @@ -0,0 +1,51 @@ +/** + * `nextly_rbac_epoch` — dialect-aware barrel. + * + * @module schemas/rbac-epoch + */ + +import type { SupportedDialect } from "@nextlyhq/adapter-drizzle/types"; + +import { NextlyError } from "../../errors/nextly-error"; + +import * as my from "./mysql"; +import * as pg from "./postgres"; +import * as sl from "./sqlite"; + +export { pg, my, sl }; + +// Re-exported from the leaf that declares them, so this barrel stays one of the +// readers rather than becoming a second author. See `./table-name`. +export { RBAC_EPOCH_TABLE, RBAC_EPOCH_ROW_ID } from "./table-name"; + +/** + * The ONE place a dialect is turned into an epoch table. + * + * A ternary chain ending in a bare `else` would assign every future dialect to + * whichever branch came last, so adding one would compile and hand back another + * dialect's table. The `never` assignment makes the compiler demand a case. + */ +function epochForDialect(dialect: SupportedDialect) { + switch (dialect) { + case "postgresql": + return pg.nextlyRbacEpoch; + case "mysql": + return my.nextlyRbacEpoch; + case "sqlite": + return sl.nextlyRbacEpoch; + default: { + const _exhaustive: never = dialect; + throw NextlyError.internal({ + logContext: { + reason: "no rbac epoch table for this dialect", + dialect: String(_exhaustive), + }, + }); + } + } +} + +/** The epoch table for the requested dialect, as a schema fragment. */ +export function rbacEpochTables(dialect: SupportedDialect) { + return { nextlyRbacEpoch: epochForDialect(dialect) }; +} diff --git a/packages/nextly/src/schemas/rbac-epoch/mysql.ts b/packages/nextly/src/schemas/rbac-epoch/mysql.ts new file mode 100644 index 0000000000..58b9c9e293 --- /dev/null +++ b/packages/nextly/src/schemas/rbac-epoch/mysql.ts @@ -0,0 +1,19 @@ +/** + * `nextly_rbac_epoch` — how many times RBAC has changed, MySQL. + * + * See `./postgres.ts` for what the table is, why it holds one row under a fixed + * key, and why it is a table rather than a column on the cache rows. + * + * @module schemas/rbac-epoch/mysql + */ + +import { bigint, mysqlTable, timestamp, varchar } from "drizzle-orm/mysql-core"; + +import { RBAC_EPOCH_TABLE } from "./table-name"; + +export const nextlyRbacEpoch = mysqlTable(RBAC_EPOCH_TABLE, { + id: varchar("id", { length: 32 }).primaryKey(), + revision: bigint("revision", { mode: "number" }).notNull(), + generation: varchar("generation", { length: 64 }).notNull(), + updatedAt: timestamp("updated_at").notNull(), +}); diff --git a/packages/nextly/src/schemas/rbac-epoch/postgres.ts b/packages/nextly/src/schemas/rbac-epoch/postgres.ts new file mode 100644 index 0000000000..4cb2a8dc08 --- /dev/null +++ b/packages/nextly/src/schemas/rbac-epoch/postgres.ts @@ -0,0 +1,61 @@ +/** + * `nextly_rbac_epoch` — how many times RBAC has changed, PostgreSQL. + * + * One row, holding a counter that rises whenever a role, a role's permissions + * or a permission row changes. Every cached authorization answer is filed under + * the value it was computed at, and an answer filed under an older value is not + * served. + * + * ## Why this is in the database rather than in memory + * + * The counter it replaces lived in a module variable, so it only ever moved in + * the process that handled the change. A second instance neither saw that move + * nor had one of its own, and went on serving what it had cached until the + * entry aged out — which for the shared tier meant a revoked grant could + * outlive its revocation by the whole TTL. A number every instance reads is the + * only thing that makes "this answer is stale" a fact about the install rather + * than about one process. + * + * ## Why one row with a fixed key + * + * `id` is always `global`. Making it the primary key means a second row cannot + * be inserted, so "there is exactly one counter" is a property of the schema + * rather than something the code has to keep true. The increment is then a + * single statement the database evaluates itself — `SET revision = revision + 1` + * — which is atomic under concurrent writers on every dialect, where a + * read-modify-write from the application would lose one of two simultaneous + * invalidations. + * + * ## Why a table rather than a column on the cache rows + * + * `ensureCoreTables` reconciles an existing database by re-running idempotent + * `CREATE TABLE IF NOT EXISTS` statements, so a NEW TABLE reaches installs that + * already exist. It explicitly does not repair a table whose columns drifted, + * which a new column on `user_permission_cache` would have needed. A separate + * table is therefore the shape that can actually be delivered to the databases + * this is meant to protect. + * + * @module schemas/rbac-epoch/postgres + */ + +import { bigint, pgTable, text, timestamp } from "drizzle-orm/pg-core"; + +import { RBAC_EPOCH_TABLE } from "./table-name"; + +export const nextlyRbacEpoch = pgTable(RBAC_EPOCH_TABLE, { + id: text("id").primaryKey(), + // `bigint` rather than `integer`: this only ever rises, and an install that + // wrapped a 32-bit counter would start serving answers filed under a value + // the counter is about to reach again. + revision: bigint("revision", { mode: "number" }).notNull(), + // Identity of the counter itself, generated once with the row. + // + // A number alone cannot tell "the same counter, unchanged" from "a different + // counter that happens to read the same" — which is what a restored backup, a + // re-provisioned environment or a failover that lost writes produces. An + // instance holding entries filed at 7 would find a fresh counter also at 7 + // and go on serving them. Comparing the pair makes a replaced store retire + // every cached answer, whatever its number says. + generation: text("generation").notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull(), +}); diff --git a/packages/nextly/src/schemas/rbac-epoch/sqlite.ts b/packages/nextly/src/schemas/rbac-epoch/sqlite.ts new file mode 100644 index 0000000000..a0439179d8 --- /dev/null +++ b/packages/nextly/src/schemas/rbac-epoch/sqlite.ts @@ -0,0 +1,20 @@ +/** + * `nextly_rbac_epoch` — how many times RBAC has changed, SQLite. + * + * See `./postgres.ts` for what the table is, why it holds one row under a fixed + * key, and why it is a table rather than a column on the cache rows. + * + * @module schemas/rbac-epoch/sqlite + */ + +import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; + +import { RBAC_EPOCH_TABLE } from "./table-name"; + +export const nextlyRbacEpoch = sqliteTable(RBAC_EPOCH_TABLE, { + id: text("id").primaryKey(), + revision: integer("revision").notNull(), + generation: text("generation").notNull(), + // Unix seconds, as every other timestamp in this dialect's tables. + updatedAt: integer("updated_at", { mode: "timestamp" }).notNull(), +}); diff --git a/packages/nextly/src/schemas/rbac-epoch/table-name.ts b/packages/nextly/src/schemas/rbac-epoch/table-name.ts new file mode 100644 index 0000000000..700037dd67 --- /dev/null +++ b/packages/nextly/src/schemas/rbac-epoch/table-name.ts @@ -0,0 +1,32 @@ +/** + * What the RBAC epoch table is called, and which row holds it. + * + * A leaf on purpose: it imports nothing, so the three dialect declarations, the + * SQLite bootstrap DDL, the core-table manifest and the runtime queries can all + * read the name from here without any of them importing each other. + * + * That matters more for this table than for most. The manifest decides which + * tables reconciliation creates and which ones introspection expects to find, + * the DDL decides what is actually created, and the runtime queries decide what + * is read — so a name spelled separately in each can be renamed in some of them + * and not others, and the result is not an error. It is a second counter: the + * install creates and reconciles one table while every epoch read and write + * goes to another, and the two never disagree loudly because nothing compares + * them. Every cache in the install then answers from a counter nothing bumps. + * + * @module schemas/rbac-epoch/table-name + */ + +/** + * The physical table name, spelled once and read everywhere. + */ +export const RBAC_EPOCH_TABLE = "nextly_rbac_epoch"; + +/** + * The key of the single row. + * + * A constant rather than a parameter: the primary key is what makes a second + * counter unrepresentable, and a caller free to choose the key could create one + * whose bumps nothing else reads. + */ +export const RBAC_EPOCH_ROW_ID = "global"; diff --git a/packages/nextly/src/services/lib/epoch-has-one-answer.test.ts b/packages/nextly/src/services/lib/epoch-has-one-answer.test.ts new file mode 100644 index 0000000000..caafb172ab --- /dev/null +++ b/packages/nextly/src/services/lib/epoch-has-one-answer.test.ts @@ -0,0 +1,148 @@ +import { readdirSync, readFileSync } from "node:fs"; +import { dirname, join, relative, sep } from "node:path"; +import { fileURLToPath } from "node:url"; + +import ts from "typescript"; +import { describe, expect, it } from "vitest"; + +/** + * The epoch answers two questions, and each has exactly one place it is asked. + * + * Both guards here exist because the same mistake was made twice in the same + * change, in a shape no test could see: a value declared as the single source + * of something, and then not actually used as one. + * + * The first is "is this stamp still current". Three tiers asked it — the + * in-memory caches, the shared tier's write gate, and the API key's copied + * grants — and two of them compared the stamp without asking whether the epoch + * was worth comparing against. Those two look identical to the correct one at a + * glance, and the difference only shows on an install whose epoch table is not + * yet reconciled, where the stamp being matched is one this process invented + * and no other instance has ever seen. + * + * The second is what the table is called. Five places spelled it, and the + * constant documented as the single spelling was read by none of them, so a + * rename could move the manifest and the DDL while leaving the runtime queries + * pointed at the old name — two tables, no error, and every cache in the + * install answering from a counter nothing bumps. + */ +const SRC = join(dirname(fileURLToPath(import.meta.url)), "..", ".."); + +/** Where each answer is allowed to be authored. */ +const EPOCH_MODULE = join("services", "lib", "rbac-epoch.ts"); +const TABLE_NAME_MODULE = join("schemas", "rbac-epoch", "table-name.ts"); + +function sourceFiles(dir: string): string[] { + return readdirSync(dir, { withFileTypes: true }).flatMap(entry => { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + return entry.name === "__tests__" ? [] : sourceFiles(full); + } + return entry.name.endsWith(".ts") && + !entry.name.endsWith(".d.ts") && + !/\.(test|integration\.test)(-d)?\.ts$/.test(entry.name) + ? [full] + : []; + }); +} + +const FILES = sourceFiles(SRC); + +function parse(file: string): ts.SourceFile { + return ts.createSourceFile( + file, + readFileSync(file, "utf8"), + ts.ScriptTarget.Latest, + true + ); +} + +/** Is this expression a call to `currentEpoch()`? */ +function isCurrentEpochCall(node: ts.Node): boolean { + return ( + ts.isCallExpression(node) && + ts.isIdentifier(node.expression) && + node.expression.text === "currentEpoch" + ); +} + +/** Every `x === currentEpoch()` (or `!==`, either way round) in one file. */ +function stampComparisons(source: ts.SourceFile): number[] { + const found: number[] = []; + const visit = (node: ts.Node): void => { + if ( + ts.isBinaryExpression(node) && + (node.operatorToken.kind === ts.SyntaxKind.EqualsEqualsEqualsToken || + node.operatorToken.kind === + ts.SyntaxKind.ExclamationEqualsEqualsToken) && + (isCurrentEpochCall(node.left) || isCurrentEpochCall(node.right)) + ) { + found.push(source.getLineAndCharacterOfPosition(node.pos).line + 1); + } + ts.forEachChild(node, visit); + }; + visit(source); + return found; +} + +describe("whether a stamp is current is asked in one place", () => { + it("is not compared against `currentEpoch()` anywhere else", () => { + // A comparison written out again is a second answer to the question + // `stampIsCurrent` exists to give, and the half it drops is the trust + // check — which is invisible until the epoch table is missing. + const offenders = FILES.flatMap(file => { + const where = relative(SRC, file); + if (where === EPOCH_MODULE) return []; + return stampComparisons(parse(file)).map( + line => `${where.split(sep).join("/")}:${line}` + ); + }); + + expect(offenders).toEqual([]); + }); + + it("finds one when there is one, so the case above is not vacuous", () => { + // The control. An AST walk that matches nothing — a renamed function, a + // visitor that never descends, a file list that resolved to an empty + // directory — satisfies the assertion above perfectly. + const written = ts.createSourceFile( + "probe.ts", + "const ok = entry.epoch === currentEpoch();\n" + + "const no = currentEpoch() !== other;\n", + ts.ScriptTarget.Latest, + true + ); + + expect(stampComparisons(written)).toHaveLength(2); + }); + + it("reads a population, so an empty file list cannot pass it", () => { + // The other half of the control: the assertion above is equally satisfied + // by having read nothing at all. + expect(FILES.length).toBeGreaterThan(500); + expect(FILES.some(f => relative(SRC, f) === EPOCH_MODULE)).toBe(true); + }); +}); + +describe("the epoch table is named in one place", () => { + it("is spelled literally nowhere but the module that declares it", () => { + const offenders = FILES.filter( + file => + relative(SRC, file) !== TABLE_NAME_MODULE && + readFileSync(file, "utf8").includes('"nextly_rbac_epoch"') + ).map(file => relative(SRC, file).split(sep).join("/")); + + expect(offenders).toEqual([]); + }); + + it("is spelled once where it IS declared, so the search can find it", () => { + // The control. Searching for a string that occurs nowhere at all reports + // the same clean result as a name that is genuinely centralised. + const declaring = FILES.filter( + file => relative(SRC, file) === TABLE_NAME_MODULE + ); + + expect(declaring).toHaveLength(1); + expect(readFileSync(declaring[0], "utf8")).toContain('"nextly_rbac_epoch"'); + }); +}); diff --git a/packages/nextly/src/services/lib/permissions-executor-cache.integration.test.ts b/packages/nextly/src/services/lib/permissions-executor-cache.integration.test.ts index 1ffeed6d7e..9362b1b8f2 100644 --- a/packages/nextly/src/services/lib/permissions-executor-cache.integration.test.ts +++ b/packages/nextly/src/services/lib/permissions-executor-cache.integration.test.ts @@ -19,7 +19,8 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { getDialectTables } from "../../database/index"; -import { hasPermission } from "./permissions"; +import { hasPermission, isSuperAdmin } from "./permissions"; +import { resetEpochForTests } from "./rbac-epoch"; let harness: TestNextly | undefined; @@ -106,3 +107,72 @@ describe("permission cache — transaction executor bypass (integration)", () => expect(await hasPermission(userId, "read", "posts")).toBe(true); }); }); + +/** + * An executor-backed check must reach its answer WITHOUT the pooled connection. + * + * The caller passing an executor is inside its own open transaction. Where the + * pool holds a single connection that transaction is holding it, so any query + * this check issues on the pool waits for a connection that cannot be released + * until the check returns: the write deadlocks and both sides wait out the + * pool timeout. `domains/collections/__tests__/publish-enforcement-pool-reentry.integration.test.ts` + * reproduces the deadlock itself, but needs a real `pool.max = 1` Postgres and + * self-skips without one. What is portable, and what actually regresses, is the + * cause: a pooled checkout taken from inside the transaction at all. + */ +describe("an executor-backed check and the connection pool", () => { + /** Count checkouts of the pooled connection, leaving the adapter otherwise real. */ + function countPooledCheckouts(): { + count: () => number; + restore: () => void; + } { + const adapter = harness!.adapter as unknown as Record; + const own = Object.getOwnPropertyDescriptor(adapter, "getDrizzle"); + const original = ( + adapter.getDrizzle as (...args: unknown[]) => unknown + ).bind(harness!.adapter); + let checkouts = 0; + adapter.getDrizzle = (...args: unknown[]) => { + checkouts += 1; + return original(...args); + }; + return { + count: () => checkouts, + restore: () => { + if (own) Object.defineProperty(adapter, "getDrizzle", own); + else delete adapter.getDrizzle; + }, + }; + } + + it("takes no pooled connection, and a pooled check proves the count is live", async () => { + const userId = "pool-exec-user"; + // Obtained BEFORE the count starts: this stands in for the executor a + // caller already holds from its own transaction. + const executor = harness!.adapter.getDrizzle(); + // The epoch read is rate-limited to one a second, so an unreset module + // would skip it and the check would take no connection either way. + resetEpochForTests(); + + const pool = countPooledCheckouts(); + try { + await hasPermission(userId, "read", "posts", executor); + await isSuperAdmin(userId, executor); + expect(pool.count(), "executor-backed checks").toBe(0); + } finally { + pool.restore(); + } + + // The control. Without it, "took no connection" is equally satisfied by a + // counter that never increments, and by a check that stopped reading the + // database at all. + resetEpochForTests(); + const pooled = countPooledCheckouts(); + try { + await hasPermission(userId, "read", "posts"); + expect(pooled.count(), "pooled check").toBeGreaterThan(0); + } finally { + pooled.restore(); + } + }); +}); diff --git a/packages/nextly/src/services/lib/permissions.ts b/packages/nextly/src/services/lib/permissions.ts index 08d0e4c014..2619940b5a 100644 --- a/packages/nextly/src/services/lib/permissions.ts +++ b/packages/nextly/src/services/lib/permissions.ts @@ -26,6 +26,13 @@ import { NextlyError } from "../../errors/nextly-error"; import { getAuthLogger } from "../../lib/logger"; import type { Logger } from "../shared"; +import { + bumpEpoch, + currentEpoch, + refreshEpoch, + stampIsCurrent, +} from "./rbac-epoch"; + if (typeof window !== "undefined") { throw new Error( "[nextly] Direct API permissions module loaded in a browser context. " + @@ -135,7 +142,7 @@ function storeSharedDecision( resource: string; allowed: boolean; roleIds: string[]; - resolvedUnder: number; + resolvedUnder: string; } ): void { const { userId, action, resource, allowed, roleIds, resolvedUnder } = @@ -149,6 +156,12 @@ function storeSharedDecision( allowed, roleIds ); + // Forced, not throttled. This is the one place the interval must not + // apply: the window being closed is the upsert's own flight time, and a + // revocation that landed inside it is by definition newer than the last + // read. Asking the cached value here would accept the write the check + // exists to catch. + await refreshEpoch({ force: true }); if (!resolvedUnderCurrentRevision(resolvedUnder)) { await service.invalidateByUser(userId); } @@ -220,9 +233,22 @@ class PermissionChecker { // Captured before the reads below; see `resolvedUnderCurrentRevision`. Both // tiers written at the end of this method are subject to the same race, and // the database tier is the worse of the two: it is shared across instances - // and its entries live for a day, so a stale decision written after a + // and its entries live longest, so a stale decision written after a // tombstone outlives everything else here. - const resolvedUnder = rbacRevisionCounter; + // + // Refreshed first, so what is captured accounts for a change made by + // ANOTHER instance. Rate-limited to one read a second, which is what makes + // this affordable on a path taken by every request, and what bounds how + // long this process can be unaware of somebody else's revocation. + // + // For a POOLED check only. The refresh is itself a pooled query, so issuing + // it from inside a caller's still-open transaction asks the pool for a + // second connection — and where the pool holds one connection, the caller's + // transaction is holding it, so the query never runs and the check never + // answers. It would also buy nothing: an executor-backed check is not + // cacheable at any tier, so every use of `resolvedUnder` below is already + // behind `!executor` and the value is never consulted. + const resolvedUnder = executor ? currentEpoch() : await refreshEpoch(); // Skip EVERY cache tier when a transaction executor is supplied. Such a check // reads through the caller's still-open (uncommitted) transaction, so its @@ -230,70 +256,20 @@ class PermissionChecker { // caches: if that transaction rolls back, a grant or denial that never // committed would otherwise be reused by later, non-transactional requests // for the cache TTL. An executor-backed check always computes fresh below. + // + // The stored tier is skipped for a second reason as well: its lookup is a + // pooled query, so making it from inside the caller's transaction is the + // same re-entry the epoch refresh above avoids. if (!executor) { - // Tier 1: In-memory instance cache (ultra-fast <1ms) - const cached = this.memo.get(key); - if (typeof cached === "boolean") return cached; - - // Tier 1b: Process-wide LRU cache (<1ms) - const hit = cache.get(key); - if (hit) { - if (hit.expiresAt > Date.now()) { - this.memo.set(key, hit.value); - // refresh LRU by deleting+setting - cache.delete(key); - cache.set(key, hit); - return hit.value; - } - // expired -> clear reverse maps - cache.delete(key); - const rids = keyToRoleIds.get(key); - keyToRoleIds.delete(key); - if (rids) for (const rid of rids) roleIdToKeys.get(rid)?.delete(key); - userIdToKeys.get(userId)?.delete(key); - } - } + const remembered = this.servedFromMemory(key); + if (typeof remembered === "boolean") return remembered; - // Tier 2: Database cache (fast ~3-5ms). Skipped when a transaction executor - // is supplied: this lookup is itself a pooled query, so running it inside the - // caller's transaction would re-enter the pool (and by the rule above must - // not serve a cached decision to a transaction-scoped check anyway). - if (this.cacheService && !executor) { - try { - const dbCached = await this.cacheService.getCachedPermission( - userId, - action, - resource - ); - // The lookup is itself awaited, so an invalidation can land while it - // is outstanding: the row it returns was read before the change and - // promoting it would put a retired decision back into tier 1 for that - // tier's whole life, having just been tombstoned in tier 2. Recompute - // instead, which is what a miss would have done anyway. - // - // NOT covered by a test, and said here rather than left to look like - // coverage: `setCachedPermission` does not take effect under - // `createTestNextly` — the table is created and a write followed by a - // read returns null — so no test can reach this branch. The predicate - // it uses is covered; this call site is not. - if (dbCached !== null && resolvedUnderCurrentRevision(resolvedUnder)) { - // Cache hit - promote to tier 1 - this.memo.set(key, dbCached); - setCacheEntry(key, dbCached, userId, []); - return dbCached; - } - } catch (error) { - // Log but don't fail - fall through to fresh computation - getAuthLogger()?.log?.("warn", { - category: "auth", - op: "cache", - message: "DB cache lookup failed, falling back to fresh computation", - userId, - action, - resource, - error: String(error), - }); - } + const stored = await this.servedFromSharedTier( + key, + { userId, action, resource }, + resolvedUnder + ); + if (typeof stored === "boolean") return stored; } // Tier 3: Fresh computation (~10ms) @@ -376,6 +352,85 @@ class PermissionChecker { } } + /** + * Tiers 1 and 1b: the per-checker memo and the process-wide LRU. + * + * `undefined` means no answer, which is not the same as `false`: a denial is + * cached on the same terms as a grant, so the two have to stay tellable + * apart all the way back to the caller. + */ + private servedFromMemory(key: string): boolean | undefined { + const memoed = this.memo.get(key); + if (typeof memoed === "boolean") return memoed; + + const hit = cache.get(key); + if (!hit) return undefined; + + if (servable(hit)) { + this.memo.set(key, hit.value); + // refresh LRU by deleting+setting + cache.delete(key); + cache.set(key, hit); + return hit.value; + } + + forgetKey(key); + return undefined; + } + + /** + * Tier 2: the stored row, promoted into tier 1 when it may still be trusted. + * + * `undefined` means no usable answer, for either reason: nothing stored, or + * something stored that an invalidation has overtaken. + */ + private async servedFromSharedTier( + key: string, + check: { userId: string; action: string; resource: string }, + resolvedUnder: string + ): Promise { + if (!this.cacheService) return undefined; + const { userId, action, resource } = check; + + try { + const dbCached = await this.cacheService.getCachedPermission( + userId, + action, + resource + ); + // The lookup is itself awaited, so an invalidation can land while it + // is outstanding: the row it returns was read before the change and + // promoting it would put a retired decision back into tier 1 for that + // tier's whole life, having just been tombstoned in tier 2. Recompute + // instead, which is what a miss would have done anyway. + // + // NOT covered by a test, and said here rather than left to look like + // coverage: `setCachedPermission` does not take effect under + // `createTestNextly` — the table is created and a write followed by a + // read returns null — so no test can reach this branch. The predicate + // it uses is covered; this call site is not. + if (dbCached === null || !resolvedUnderCurrentRevision(resolvedUnder)) { + return undefined; + } + // Cache hit - promote to tier 1 + this.memo.set(key, dbCached); + setCacheEntry(key, dbCached, userId, []); + return dbCached; + } catch (error) { + // Log but don't fail - fall through to fresh computation + getAuthLogger()?.log?.("warn", { + category: "auth", + op: "cache", + message: "DB cache lookup failed, falling back to fresh computation", + userId, + action, + resource, + error: String(error), + }); + return undefined; + } + } + async hasAnyPermission( userId: string, checks: PermissionCheck[] @@ -520,7 +575,27 @@ class PermissionChecker { } // ---- Process-wide LRU cache with TTL ---- -type CacheValue = { value: boolean; expiresAt: number }; +type CacheValue = { value: boolean; expiresAt: number; epoch: string }; + +/** + * May this entry still be SERVED? + * + * Gating the write alone is not enough, and that is the half this was missing. + * A write filed under the current epoch is correct at the moment it happens; + * what retires it is the epoch moving afterwards, which is exactly the case a + * change on ANOTHER instance produces — nothing local clears the entry, and + * without this it is served until it expires. + * + * Asked in one place so the two tiers cannot answer differently, and so a third + * one added later has somewhere obvious to ask. + */ +function servable(entry: { expiresAt: number; epoch: string }): boolean { + // Derived rather than restated. Whether a stamp is still current is one + // question with one answer, and the tiers that asked it separately did not + // all remember that a stamp is only worth comparing while the epoch is + // trustworthy. + return stampIsCurrent(entry.epoch) && entry.expiresAt > Date.now(); +} const cacheTtlMs = 60_000; // 60 seconds // Memory cache size: configurable via PERMISSION_CACHE_MEMORY_SIZE env var const cacheMaxEntries = @@ -530,6 +605,27 @@ const keyToRoleIds = new Map>(); const roleIdToKeys = new Map>(); const userIdToKeys = new Map>(); +/** + * Drop one key and every index that points at it. + * + * Four callers evict a key for four different reasons — the entry expired, the + * LRU is full, the user changed, the role changed — and every one of them has + * to unpick the same three reverse maps. Written out per caller, an index left + * behind holds a key nothing can reach: the next invalidation for that user or + * role iterates a name that is no longer in the cache and skips the entry it + * was raised to remove. + */ +function forgetKey(key: string): void { + cache.delete(key); + const roleIds = keyToRoleIds.get(key); + keyToRoleIds.delete(key); + if (roleIds) for (const rid of roleIds) roleIdToKeys.get(rid)?.delete(key); + const owner = key.split("|", 1)[0]; + const owned = userIdToKeys.get(owner); + owned?.delete(key); + if (owned?.size === 0) userIdToKeys.delete(owner); +} + function setCacheEntry( key: string, value: boolean, @@ -539,17 +635,13 @@ function setCacheEntry( // simple eviction of oldest if (cache.size >= cacheMaxEntries) { const oldest = cache.keys().next().value; - if (oldest) { - cache.delete(oldest); - const rids = keyToRoleIds.get(oldest); - keyToRoleIds.delete(oldest); - if (rids) for (const rid of rids) roleIdToKeys.get(rid)?.delete(oldest); - const u = oldest.split("|", 1)[0]; - userIdToKeys.get(u)?.delete(oldest); - if (userIdToKeys.get(u)?.size === 0) userIdToKeys.delete(u); - } + if (oldest) forgetKey(oldest); } - cache.set(key, { value, expiresAt: Date.now() + cacheTtlMs }); + cache.set(key, { + value, + expiresAt: Date.now() + cacheTtlMs, + epoch: currentEpoch(), + }); const roleSet = new Set(roleIds); keyToRoleIds.set(key, roleSet); for (const rid of roleSet) { @@ -699,7 +791,10 @@ export async function listEffectivePermissions( * couple of indexed queries; a stale grant is the whole catalogue in the hands * of somebody who no longer holds the role that granted it. */ -let rbacRevisionCounter = 0; +// Kept as a NAME rather than a variable. The count now lives in +// `rbac-epoch`, where every instance can read it; leaving a second copy here +// would be two answers to one question, and the local one would win every +// comparison it took part in. /** * How many retirements are currently emptying the caches. @@ -712,8 +807,8 @@ let rbacRevisionCounter = 0; let permissionFlushDepth = 0; /** The current count; see {@link invalidatePermissionCache}. */ -export function rbacRevision(): number { - return rbacRevisionCounter; +export function rbacRevision(): string { + return currentEpoch(); } /** @@ -739,8 +834,13 @@ export function rbacRevision(): number { * and promotes the retired answer into a tier that outlives the retirement. * Nothing is cacheable while the caches are being emptied. */ -export function resolvedUnderCurrentRevision(revision: number): boolean { - return permissionFlushDepth === 0 && revision === rbacRevisionCounter; +export function resolvedUnderCurrentRevision(revision: string): boolean { + // Same derivation as `servable`, and it was missing here. Comparing the + // stamp alone accepts a result resolved under an epoch this process invented + // while the shared row was unreachable — and the forced refresh that was + // meant to close that window fails on exactly the installs where the row is + // unreachable, so the comparison is against the same unmoved value. + return permissionFlushDepth === 0 && stampIsCurrent(revision); } /** @@ -753,7 +853,7 @@ export function resolvedUnderCurrentRevision(revision: number): boolean { */ const superAdminCache = new Map< string, - { value: boolean; expiresAt: number } + { value: boolean; expiresAt: number; epoch: string } >(); const SUPER_ADMIN_CACHE_TTL_MS = 60_000; // 60 seconds @@ -825,10 +925,30 @@ export async function inPermissionSweep(run: () => Promise): Promise { try { return await permissionSweep.run(batch, run); } finally { - if (batch.dirty) await flushPermissionCaches(); + if (batch.dirty) { + // Released before the flush, which raises it again for its own window. + permissionFlushDepth -= 1; + await flushPermissionCaches(); + } } } +/** + * Empty the tiers this process holds in memory. + * + * Free, unlike the stored tier: no query, no lock, nothing to defer. So it + * happens on every invalidation including the ones a batch defers the expensive + * half of, and it is written once because four callers emptying five maps by + * hand is four chances to forget the fifth. + */ +function clearInMemoryTiers(): void { + cache.clear(); + keyToRoleIds.clear(); + roleIdToKeys.clear(); + userIdToKeys.clear(); + superAdminCache.clear(); +} + export async function invalidateAllPermissionCaches(): Promise { // Inside a sweep the caches are retired once, at the end. The revision still // advances immediately, so nothing in flight can file a result as current @@ -839,8 +959,26 @@ export async function invalidateAllPermissionCaches(): Promise { // caches now, however long the batch still has to run. const batch = permissionSweep.getStore(); if (batch) { - batch.dirty = true; - rbacRevisionCounter += 1; + if (!batch.dirty) { + batch.dirty = true; + // Held from the batch's first write to its exit. Nothing may be filed as + // current while a batch is open: its stored rows are deliberately still + // live, so a resolution that finished inside the batch would be caching + // an answer the batch has already invalidated. + permissionFlushDepth += 1; + } + // Retired here, announced at the exit. The tiers this process holds cost + // nothing to empty, so an answer it is already holding does not outlive the + // row it came from; what the batch defers is the unfiltered rewrite of + // every STORED row, which is the cost it exists for. + // + // The epoch stays where it is until that rewrite has happened. Publishing + // per write would tell every other instance to drop its in-memory answer + // and read a stored row this batch has deliberately not tombstoned yet, + // filing it under the new epoch where the batch's own exit cannot reach it + // — and for as long as the batch runs, which for a seeder is not bounded by + // anything this module controls. + clearInMemoryTiers(); return; } await flushPermissionCaches(); @@ -879,38 +1017,85 @@ export async function writingPermissions( } } -async function flushPermissionCaches(): Promise { - cache.clear(); - keyToRoleIds.clear(); - roleIdToKeys.clear(); - userIdToKeys.clear(); - superAdminCache.clear(); - rbacRevisionCounter += 1; - - if (CACHE_ENABLED) { - // Held across the shared write, so nothing computed while the stored rows - // are still readable can be filed as current. Advancing the revision again - // afterwards would not do it: the window belongs to checks that both start - // and finish inside it, and those see two numbers that never moved. - permissionFlushDepth += 1; - try { - await new PermissionCacheService(getAdapter(), getLogger(), { - cacheTtlSeconds: CACHE_TTL_SECONDS, - }).invalidateAll(); - } catch (error) { - getAuthLogger()?.log?.("error", { - category: "auth", - op: "cache", - message: "DB cache invalidation failed", - error: String(error), - }); - // Don't throw - cache invalidation failures should not break operations - } finally { - permissionFlushDepth -= 1; +/** + * Empty the SHARED tier, then publish the epoch that retires everything else. + * + * That order is the invariant, not an implementation detail. The epoch is the + * only signal another instance ever receives, and it is a one-way barrier: an + * answer filed before it moves is retired by the move, an answer filed after it + * is not. + * + * Publishing first therefore opens a window on every OTHER instance, and it is + * the worst-shaped window here. That instance rejects its own in-memory answer + * because the epoch moved, falls through to the stored row the tombstone has + * not reached yet, and promotes that retired decision back into memory under + * the NEW epoch — where nothing still to happen can reach it. Nothing local + * defends against this: the depth below is one process's own, and the instance + * doing the promoting is not the one invalidating. + * + * Emptying the stored rows first closes it from both sides. An instance that + * has not yet seen the move promotes under the OLD epoch, and the move retires + * that too; one that has seen the move finds nothing left to promote. + * + * The depth is held across both steps because THIS process has the same window + * between the tombstone starting and the epoch being published; see + * {@link resolvedUnderCurrentRevision}, which is where it is read. + * + * ## A retirement that failed publishes nothing + * + * The epoch means "everything filed before this is gone", and that sentence is + * false about stored rows a failed tombstone left live. Publishing it anyway is + * worse than staying quiet, not merely unhelpful: every other instance rejects + * its own in-memory answer BECAUSE the epoch moved, falls through to a row that + * is still there, and files it under the new epoch where nothing left to happen + * can reach it. Announcing the retirement is what converts rows that would have + * aged out into copies with a fresh life. + * + * Staying quiet leaves those rows served until their own expiry, which is what + * the install had before any of this existed. The write still happened and the + * failure is still reported; what is withheld is only the claim. + */ +async function retireSharedThenPublish( + // Whatever the retirement answers is discarded: how many rows a tombstone + // touched is not a signal anything here acts on, and a driver that reports it + // differently must not become a branch. Whether it THREW is the signal. + retireShared: () => Promise, + logContext: Record = {} +): Promise { + permissionFlushDepth += 1; + try { + if (CACHE_ENABLED) { + try { + await retireShared(); + } catch (error) { + getAuthLogger()?.log?.("error", { + category: "auth", + op: "cache", + message: "DB cache invalidation failed", + ...logContext, + error: String(error), + }); + // Reported, not thrown: an invalidation failure must not break the + // write that raised it. Unpublished, though, for the reason above. + return; + } } + await bumpEpoch(); + } finally { + permissionFlushDepth -= 1; } } +async function flushPermissionCaches(): Promise { + clearInMemoryTiers(); + + await retireSharedThenPublish(() => + new PermissionCacheService(getAdapter(), getLogger(), { + cacheTtlSeconds: CACHE_TTL_SECONDS, + }).invalidateAll() + ); +} + export async function invalidatePermissionCache( _hint: { userId?: string; roleId?: string } = {} ): Promise { @@ -933,41 +1118,24 @@ export async function invalidatePermissionCache( if (userId) superAdminCache.delete(userId); if (roleId) superAdminCache.clear(); - // Anything derived from these rows is stale from here, whoever holds it. - rbacRevisionCounter += 1; - - // Invalidate in-memory caches (Tier 1) + // Invalidate in-memory caches (Tier 1). Copied before iterating, because + // dropping a key edits the very index being walked. if (userId) { - const keys = userIdToKeys.get(userId); - if (keys) { - for (const k of keys) { - cache.delete(k); - const rids = keyToRoleIds.get(k); - keyToRoleIds.delete(k); - if (rids) for (const rid of rids) roleIdToKeys.get(rid)?.delete(k); - } - userIdToKeys.delete(userId); - } + for (const key of [...(userIdToKeys.get(userId) ?? [])]) forgetKey(key); + userIdToKeys.delete(userId); } if (roleId) { - const keys = roleIdToKeys.get(roleId); - if (keys) { - for (const k of keys) { - cache.delete(k); - const rids = keyToRoleIds.get(k); - keyToRoleIds.delete(k); - if (rids) for (const rid of rids) roleIdToKeys.get(rid)?.delete(k); - const uid = k.split("|", 1)[0]; - userIdToKeys.get(uid)?.delete(k); - if (userIdToKeys.get(uid)?.size === 0) userIdToKeys.delete(uid); - } - roleIdToKeys.delete(roleId); - } + for (const key of [...(roleIdToKeys.get(roleId) ?? [])]) forgetKey(key); + roleIdToKeys.delete(roleId); } - // Invalidate database cache (Tier 2) - if (CACHE_ENABLED) { - try { + // Invalidate the database tier (Tier 2), and only then announce it. Same + // ordering as `flushPermissionCaches`, for the reason stated on + // `retireSharedThenPublish`: a scoped tombstone is still an awaited write, so + // announcing first still lets another instance promote the row it is about to + // remove into a place the removal cannot reach. + await retireSharedThenPublish( + async () => { const cacheService = new PermissionCacheService( getAdapter(), getLogger(), @@ -982,18 +1150,9 @@ export async function invalidatePermissionCache( if (roleId) { await cacheService.invalidateByRole(roleId); } - } catch (error) { - getAuthLogger()?.log?.("error", { - category: "auth", - op: "cache", - message: "DB cache invalidation failed", - userId, - roleId, - error: String(error), - }); - // Don't throw - cache invalidation failures should not break operations - } - } + }, + { userId, roleId } + ); } /** @@ -1031,12 +1190,18 @@ export async function isSuperAdmin( // grants resolved from it would then be cached under the NEW revision, // putting the catalogue back for a full five minutes in the hands of somebody // who had just lost the role. - const resolvedUnder = rbacRevisionCounter; + // + // Refreshed rather than read, for the reason `hasPermission` gives: a + // demotion performed on another instance has to reach this one. And skipped + // for an executor-backed check, for the other reason `hasPermission` gives: + // the refresh is a pooled query, and the caller's transaction may be holding + // the only connection there is. + const resolvedUnder = executor ? currentEpoch() : await refreshEpoch(); // Check in-memory cache (only for pooled, committed-view checks). if (!executor) { const cached = superAdminCache.get(userId); - if (cached && cached.expiresAt > Date.now()) { + if (cached && servable(cached)) { return cached.value; } } @@ -1056,6 +1221,7 @@ export async function isSuperAdmin( superAdminCache.set(userId, { value: false, expiresAt: Date.now() + SUPER_ADMIN_CACHE_TTL_MS, + epoch: currentEpoch(), }); } return false; @@ -1085,6 +1251,7 @@ export async function isSuperAdmin( superAdminCache.set(userId, { value: result, expiresAt: Date.now() + SUPER_ADMIN_CACHE_TTL_MS, + epoch: currentEpoch(), }); // Evict oldest if cache grows too large diff --git a/packages/nextly/src/services/lib/rbac-epoch.test.ts b/packages/nextly/src/services/lib/rbac-epoch.test.ts new file mode 100644 index 0000000000..521bcfa4fd --- /dev/null +++ b/packages/nextly/src/services/lib/rbac-epoch.test.ts @@ -0,0 +1,441 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { container } from "../../di/container"; + +import { + EPOCH_TTL_MS, + bumpEpoch, + currentEpoch, + epochIsTrustworthy, + refreshEpoch, + resetEpochForTests, + stampIsCurrent, +} from "./rbac-epoch"; + +/** + * The epoch answers for the INSTALL, and keeps answering when it cannot. + * + * Two properties matter and they pull against each other. It has to reflect a + * change made by another instance, which is the whole reason it left memory. + * And it must never fail an authorization check to do so: an install upgraded + * from a version without the table has nothing to read until its core tables + * are reconciled, and refusing every request until then would be a worse + * outcome than the staleness this replaces. + * + * A fake adapter rather than a database, because what is under test is the + * read-and-cache logic rather than SQL. The statements themselves are exercised + * against real databases by the integration suite that drives invalidation end + * to end. + */ +type Row = { revision: number; generation: string }; + +function chainOf(result: unknown) { + const self: Record = { + from: () => self, + where: () => self, + set: () => self, + values: () => self, + limit: () => self, + then: (resolve: (value: unknown) => unknown) => resolve(result), + }; + return self; +} + +function fakeAdapter(rows: () => Row[], onWrite?: () => void) { + const chain = (result: unknown) => { + const self: Record = { + from: () => self, + where: () => self, + set: () => self, + values: () => self, + limit: () => self, + then: (resolve: (value: unknown) => unknown) => resolve(result), + }; + return self; + }; + return { + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => ({ + select: () => chain(rows()), + update: () => { + onWrite?.(); + return chain([{ changes: 1 }]); + }, + insert: () => chain([{ changes: 1 }]), + }), + }; +} + +/** + * A shared row that actually applies the increment the statement carries. + * + * `fakeAdapter` answers reads and reports writes as successful without moving + * anything, which cannot show a count applied twice or lost. The upsert's own + * `values({ revision })` carries the backlog being claimed, so the fake adds + * exactly what the statement asked the database to add. + */ +function countingAdapter(state: { revision: number; generation: string }) { + return { + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => ({ + select: () => + chainOf([{ revision: state.revision, generation: state.generation }]), + insert: () => { + let owed = 0; + const self: Record = { + values: (row: { revision: number }) => { + owed = Number(row.revision); + return self; + }, + onConflictDoUpdate: () => { + state.revision += owed; + return Promise.resolve([{ changes: 1 }]); + }, + then: (resolve: (value: unknown) => unknown) => + resolve([{ changes: 1 }]), + }; + return self; + }, + update: () => chainOf([{ changes: 1 }]), + }), + }; +} + +/** A select chain whose answer arrives when the given promise settles. */ +function pending(answer: Promise) { + const self: Record = { + from: () => self, + where: () => self, + limit: () => self, + then: ( + resolve: (value: Row[]) => unknown, + reject: (reason: unknown) => unknown + ) => answer.then(resolve, reject), + }; + return self; +} + +function install(adapter: unknown) { + container.register("adapter", () => adapter); +} + +describe("the RBAC epoch answers for the install", () => { + beforeEach(() => { + resetEpochForTests(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + resetEpochForTests(); + }); + + it("starts at zero and reads the shared value", async () => { + install(fakeAdapter(() => [{ revision: 7, generation: "g" }])); + + expect(currentEpoch()).toBe(":0"); + await expect(refreshEpoch()).resolves.toBe("g:7"); + expect(currentEpoch()).toBe("g:7"); + }); + + it("takes another instance's bump on the next refresh", async () => { + // The property the whole module exists for: a change this process did not + // make still retires what it has cached. + let shared = 3; + install(fakeAdapter(() => [{ revision: shared, generation: "g" }])); + + await refreshEpoch(); + expect(currentEpoch()).toBe("g:3"); + + shared = 4; + vi.advanceTimersByTime(EPOCH_TTL_MS); + await refreshEpoch(); + + expect(currentEpoch()).toBe("g:4"); + }); + + it("reads at most once per interval, so the hot path is not a query", async () => { + // The control on the case above. Without it, "it sees the new value" is + // satisfied by reading on every single check, which is the cost this design + // exists to avoid. + let reads = 0; + install( + fakeAdapter(() => { + reads += 1; + return [{ revision: 1, generation: "g" }]; + }) + ); + + await refreshEpoch(); + await refreshEpoch(); + await refreshEpoch(); + + expect(reads).toBe(1); + }); + + it("answers only with values the shared row gave it", async () => { + // The property that removes a whole class of bug. Inventing `epoch + 1` + // locally means two counters that both advance, and the local one then + // wins every comparison — so an instance that bumped while the shared row + // was unreachable would stay permanently ahead and stop noticing anybody + // else. Whatever the row says is what this answers. + let shared = 9; + install(fakeAdapter(() => [{ revision: shared, generation: "g" }])); + await refreshEpoch(); + expect(currentEpoch()).toBe("g:9"); + + // A bump the row does not reflect must not invent a value: the fake's read + // still answers 9, so 9 is what this process may claim. + shared = 9; + await bumpEpoch(); + + expect(currentEpoch()).toBe("g:9"); + }); + + it("distrusts its caches while an invalidation is owed", async () => { + // The fail-safe direction. An epoch other instances have never seen cannot + // decide whether an answer is current, so nothing may be served from cache + // until the shared row accepts the change. + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => { + throw new Error("no such table: nextly_rbac_epoch"); + }, + }); + + expect(epochIsTrustworthy()).toBe(true); + await bumpEpoch(); + expect(epochIsTrustworthy()).toBe(false); + }); + + it("trusts them again once the shared row accepts the backlog", async () => { + // The control on the case above, and the recovery path. Without it, + // "distrusts while owed" is satisfied by never trusting anything again, + // which would leave an install permanently uncached after one hiccup. + let reachable = false; + let shared = 0; + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => { + if (!reachable) throw new Error("unreachable"); + return { + select: () => chainOf([{ revision: shared, generation: "g" }]), + // The real statement is an upsert, so the fake has to be one too: a + // fake that only models the UPDATE reports a backlog as persisted + // while the counter never moves. + insert: () => { + const chain = chainOf([{ changes: 1 }]) as Record; + chain.values = () => chain; + chain.onConflictDoUpdate = () => { + shared += 1; + return Promise.resolve([{ changes: 1 }]); + }; + return chain; + }, + update: () => chainOf([{ changes: 1 }]), + }; + }, + }); + + await bumpEpoch(); + expect(epochIsTrustworthy()).toBe(false); + + reachable = true; + vi.advanceTimersByTime(EPOCH_TTL_MS); + await refreshEpoch(); + + expect(epochIsTrustworthy()).toBe(true); + // And the invalidation made while unreachable reached the row rather than + // being dropped on the way. + expect(shared).toBeGreaterThan(0); + }); + + it("collapses concurrent refreshes onto one read", async () => { + // The interval bounds when a read may START, not how many run. Every check + // arriving after expiry sees the same stale timestamp, so without sharing + // the in-flight promise a burst issues one query per request. + let reads = 0; + install( + fakeAdapter(() => { + reads += 1; + return [{ revision: 1, generation: "g" }]; + }) + ); + + await Promise.all([refreshEpoch(), refreshEpoch(), refreshEpoch()]); + + expect(reads).toBe(1); + }); + + it("rate-limits FAILING reads too, so a missing table is not a query storm", async () => { + // The upgrade window. Leaving the timestamp unset on failure means one + // failing query per authorization check rather than one per interval. + let attempts = 0; + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => { + attempts += 1; + throw new Error("no such table"); + }, + }); + + await refreshEpoch(); + await refreshEpoch(); + await refreshEpoch(); + + expect(attempts).toBe(1); + }); + + it("forces a read when told to, whatever the interval says", async () => { + // The post-write verification depends on this: the window it closes is the + // write's own flight time, so a revocation inside it is newer than the last + // read by definition. + let shared = 1; + install(fakeAdapter(() => [{ revision: shared, generation: "g" }])); + await refreshEpoch(); + + shared = 2; + await expect(refreshEpoch()).resolves.toBe("g:1"); + await expect(refreshEpoch({ force: true })).resolves.toBe("g:2"); + }); + + it("keeps answering when the table cannot be read", async () => { + // An install that has not reconciled its core tables has no row to read. + // Degrading to a local counter is what it had before; failing the check + // would be worse than not having upgraded at all. + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => { + throw new Error("no such table: nextly_rbac_epoch"); + }, + }); + + // Neither call throws, which is the property: an authorization check must + // not fail because a counter is unreachable. + await expect(refreshEpoch()).resolves.toBe(":0"); + await expect(bumpEpoch()).resolves.toBe(":0"); + }); + + it("still takes effect locally when the shared write fails", async () => { + // The half of degrading that matters. A shared write nobody can make must + // not stop this process retiring its own caches, or a failed upgrade turns + // a staleness bug into a revocation that never happens anywhere. + // + // It takes effect by refusing to serve rather than by moving the number. + // Moving it would invent a value no other instance has seen, which is the + // divergence this model exists to make unrepresentable. + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => { + throw new Error("unwritable"); + }, + }); + + await bumpEpoch(); + + expect(epochIsTrustworthy()).toBe(false); + }); + + it("lands both of two invalidations that race, rather than one", async () => { + // A second invalidation raised while the first is still writing has to + // reach the row on its own account. Folded into a write already in flight + // it is simply gone: no other instance is ever told about it, while this + // one goes on believing its backlog is persisted. + const shared = { revision: 0, generation: "g" }; + install(countingAdapter(shared)); + + await Promise.all([bumpEpoch(), bumpEpoch()]); + + // Two invalidations, two increments — not one, and not three. + expect(shared.revision).toBe(2); + // And nothing is left owed, which is what lets caching resume. + expect(epochIsTrustworthy()).toBe(true); + }); + + it("does not hand a forced caller a read that began before it", async () => { + // Forcing exists for the post-write verification, whose window is the + // write's own flight time. A read already running may have queried before + // that write, so joining it answers with an observation older than the + // thing being confirmed — the check then passes on evidence that predates + // what it is checking. + let shared = 1; + let reads = 0; + let releaseFirst: () => void = () => {}; + + function rows() { + // Snapshotted when the query is ISSUED, which is what makes a held read + // an old observation rather than a slow one. + const snapshot = [{ revision: shared, generation: "g" }]; + reads += 1; + if (reads > 1) return Promise.resolve(snapshot); + return new Promise(resolve => { + releaseFirst = () => resolve(snapshot); + }); + } + + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => ({ + select: () => pending(rows()), + insert: () => chainOf([{ changes: 1 }]), + update: () => chainOf([{ changes: 1 }]), + }), + }); + + const first = refreshEpoch(); + // Let that read reach the database before anything moves under it. Asserted + // rather than assumed: if it has not started, the case below proves nothing. + await Promise.resolve(); + await Promise.resolve(); + expect(reads).toBe(1); + + // The change a forced caller is verifying, made while the first read is out. + shared = 2; + const forced = refreshEpoch({ force: true }); + + releaseFirst(); + await expect(first).resolves.toBe("g:1"); + await expect(forced).resolves.toBe("g:2"); + expect(reads).toBe(2); + }); + + it("refuses a stamp that matches while an invalidation is still owed", async () => { + // The whole reason the tiers ask this rather than comparing stamps + // themselves. The stamp matches — it is the value this process is + // answering with — and it matches nothing any other instance has seen, + // because the change that produced it never reached the shared row. + install({ + getCapabilities: () => ({ dialect: "sqlite" as const }), + getDrizzle: () => { + throw new Error("no such table: nextly_rbac_epoch"); + }, + }); + + const stamp = currentEpoch(); + expect(stampIsCurrent(stamp)).toBe(true); + + await bumpEpoch(); + + expect(currentEpoch()).toBe(stamp); + expect(stampIsCurrent(stamp)).toBe(false); + }); + + it("accepts a matching stamp and refuses a stale one when nothing is owed", async () => { + // The control on both halves. Without the first, "refuses while owed" is + // satisfied by refusing everything; without the second, by accepting + // everything. + install(fakeAdapter(() => [{ revision: 4, generation: "g" }])); + await refreshEpoch(); + + expect(stampIsCurrent("g:4")).toBe(true); + expect(stampIsCurrent("g:3")).toBe(false); + }); + + it("treats a missing row as epoch zero rather than as a failure", async () => { + // A table that exists with nothing in it is a fresh install that has never + // invalidated, which is zero — not an error, and not a reason to degrade. + install(fakeAdapter(() => [])); + + await expect(refreshEpoch()).resolves.toBe(":0"); + }); +}); diff --git a/packages/nextly/src/services/lib/rbac-epoch.ts b/packages/nextly/src/services/lib/rbac-epoch.ts new file mode 100644 index 0000000000..446551c725 --- /dev/null +++ b/packages/nextly/src/services/lib/rbac-epoch.ts @@ -0,0 +1,399 @@ +/** + * The RBAC epoch: how many times authorization data has changed, install-wide. + * + * Every cache of an authorization answer is filed under the epoch it was + * computed at, and an answer filed under an older one is not served. The + * counter it replaces was a module variable, so it moved only in the process + * that handled the change: a second instance neither saw the move nor had one + * of its own, and served what it had cached until the entry aged out. + * + * ## Read from memory, refreshed on a timer + * + * `currentEpoch()` is synchronous because the checks that ask it are — they sit + * between a read and a cache write, on a path taken on every request. So the + * value is held in memory and refreshed by `refreshEpoch()`, which the async + * entry points call before they capture the epoch they will file under. + * + * The refresh is rate-limited to one read per {@link EPOCH_TTL_MS}. That + * interval is the bound on how long an instance can be unaware of another + * instance's change, and it buys back the per-request query the naive version + * would cost: one indexed read per second per instance rather than one per + * authorization check. + * + * An instance's OWN change is not subject to that delay. `bumpEpoch` advances + * the in-memory value as soon as the write lands, so the process that made the + * change never serves a stale answer of its own making. + * + * ## Degrading when the table is not there + * + * An installation upgraded from a version without this table has no row to read + * until `nextly db:sync` reconciles the core tables. Every read and write here + * therefore degrades to the previous behaviour — a counter local to this + * process — rather than failing the authorization check that asked. The + * degraded state is strictly what the install had before, so it cannot be worse + * than not having upgraded; it is reported once so an operator can see why + * cross-instance invalidation is not yet in effect. + * + * @module services/lib/rbac-epoch + */ +import { randomUUID } from "node:crypto"; + +import type { DrizzleAdapter } from "@nextlyhq/adapter-drizzle"; +import { eq, sql } from "drizzle-orm"; + +import { container } from "../../di/container"; +import { getAuthLogger } from "../../lib/logger"; +import { RBAC_EPOCH_ROW_ID, rbacEpochTables } from "../../schemas/rbac-epoch"; + +/** + * How long a read of the shared counter is reused. + * + * This is the bound on cross-instance staleness, so it is deliberately short. + * A second is long enough to collapse a burst of authorization checks onto one + * query and short enough that a revocation is everywhere before an operator has + * finished watching for it. + */ +export const EPOCH_TTL_MS = 1000; + +/** + * The epoch this process is currently answering with, and when it was read. + * + * `readAt` is zero until the first successful read, which is what makes the + * first `refreshEpoch()` of a process actually go to the database rather than + * trusting an initial value nothing established. + */ +let revision = 0; +let generation = ""; +let readAt = 0; + +/** + * A refresh already on its way, shared by every ORDINARY caller while it runs. + * + * Without this the interval bounds how often a read STARTS being allowed, not + * how many run: every check arriving after expiry sees the same stale `readAt` + * and issues its own query before any of them finishes, so a burst turns one + * read per second into one per request — the cost this design exists to avoid. + * + * A forced caller does not join it; see {@link refreshEpoch}. + */ +let inFlight: Promise | null = null; + +/** + * Local invalidations the shared row has not accepted yet. + * + * Only ever above zero while the shared store is unreachable. It is a COUNT + * rather than a second epoch on purpose: two counters that both advance + * diverge, and the local one then wins every comparison, so an instance that + * invalidated while degraded would stay permanently ahead and stop noticing + * anybody else's changes. Nothing here invents an epoch; the only values + * `epoch` ever takes are ones the shared row gave it. + */ +let pendingBumps = 0; + +/** + * Whether the shared counter has been found unreadable. + * + * Held so the warning is emitted once rather than on every check: an install + * that has not reconciled its core tables would otherwise log per request, and + * a log nobody can read is the same as no log. + */ +let degraded = false; + +/** + * The narrow view of the query builder this module uses. + * + * `getDrizzle()` is typed `unknown` because the concrete builder differs per + * driver, and the repository narrows it structurally at each call site rather + * than casting — the same shape `services/lib/permissions.ts` declares for its + * reads. Spelling out only the three statements used here keeps the module + * typed without an escape hatch, and makes a driver that stops offering one of + * them a compile error rather than a runtime one. + */ +interface EpochRow { + revision: number; + generation: string; +} +interface EpochSelect extends Promise { + from(table: unknown): EpochSelect; + where(condition: unknown): EpochSelect; + limit(count: number): EpochSelect; +} +interface EpochInsert extends Promise { + values(row: Record): EpochInsert; + // Postgres and SQLite spell the upsert one way, MySQL the other. Both are + // optional so a builder offering neither is a branch rather than a crash. + onConflictDoUpdate?: (config: { + target: unknown; + set: Record; + }) => Promise; + onDuplicateKeyUpdate?: (config: { + set: Record; + }) => Promise; +} +interface EpochUpdate extends Promise { + set(patch: Record): EpochUpdate; + where(condition: unknown): EpochUpdate; +} +interface EpochExecutor { + select(projection: Record): EpochSelect; + insert(table: unknown): EpochInsert; + update(table: unknown): EpochUpdate; +} + +function adapter(): DrizzleAdapter { + return container.get("adapter"); +} + +function executor(): EpochExecutor { + return adapter().getDrizzle(); +} + +function epochTable() { + const { dialect } = adapter().getCapabilities(); + return rbacEpochTables(dialect).nextlyRbacEpoch; +} + +function reportDegraded(error: unknown): void { + if (degraded) return; + degraded = true; + getAuthLogger()?.log?.("warn", { + category: "auth", + op: "cache", + message: + "RBAC epoch table unreadable; cache invalidation is local to this " + + "process until `nextly db:sync` reconciles the core tables", + error: String(error), + }); +} + +/** + * The epoch to file a cached answer under, without going to the database. + * + * Callers that are about to READ and then cache should call + * {@link refreshEpoch} first, so the value they file under reflects any change + * another instance made. + */ +export function currentEpoch(): string { + return `${generation}:${String(revision)}`; +} + +/** + * May a cached answer be trusted at all right now? + * + * False while this process holds invalidations the shared row has not + * accepted. Its epoch is then a value other instances have never seen, so + * comparing anything against it says nothing about whether that answer is + * current — and the honest response to "I cannot tell" on an authorization + * decision is to recompute rather than to serve. + * + * The cost lands only on an installation whose core tables are not reconciled + * AND which has since changed a role: it stops serving from cache until + * `nextly db:sync` runs. That is a visible, self-correcting slowdown rather + * than an invisible stale grant, and it is the direction to fail in. + */ +export function epochIsTrustworthy(): boolean { + return pendingBumps === 0; +} + +/** + * May an answer filed under `stamp` still be used? + * + * The one place that question is answered, because it has three askers and they + * were not asking the same thing. Comparing the stamp is only half of it: a + * match means nothing while this process holds invalidations the shared row has + * not accepted, since the value being matched against is then one no other + * instance has ever seen. A tier that compares stamps and omits the trust check + * looks correct beside one that does not, and serves a revoked answer for its + * whole life on an install whose epoch table is not yet reconciled. + * + * Every cache derives its own predicate from this rather than restating it: + * the in-memory tiers add their expiry, the shared tier adds its retirement, + * and the API key's copied grants add their own freshness window. + */ +export function stampIsCurrent(stamp: string): boolean { + if (!epochIsTrustworthy()) return false; + return stamp === currentEpoch(); +} + +/** + * Bring this process's copy of the epoch up to date, at most once per TTL. + * + * Answers the epoch it settled on, so a caller can capture it in the same + * expression rather than reading it again afterwards and racing its own + * refresh. + */ +export async function refreshEpoch(options?: { + force?: boolean; +}): Promise { + const force = options?.force === true; + if (!force && Date.now() - readAt < EPOCH_TTL_MS) return currentEpoch(); + + // Joining a read already running is honest for an ordinary caller: it asked + // for a value no older than the interval, and that read will supply one. + // + // It is not honest for a FORCED caller. The only reason to force is to + // observe something that has just happened, and a read already in flight may + // have queried before it — handing that back answers with an observation + // older than the write it exists to confirm, which is the whole of the + // post-write verification. So a forced caller waits out the reads ahead of it + // and then makes one of its own. + if (!force) { + if (inFlight) return inFlight; + } else { + // Sequential by design: each read ahead of this one has to finish before a + // new observation may start, or the observation is not a new one. + let ahead = inFlight; + while (ahead) { + await ahead; + ahead = inFlight; + } + } + + const reading = readShared().finally(() => { + if (inFlight === reading) inFlight = null; + }); + inFlight = reading; + return reading; +} + +async function readShared(): Promise { + try { + // Owed invalidations go in BEFORE the value comes out, and that order is + // the point: a read taken first answers with a number that does not + // include them, and this process would then adopt it as authoritative and + // trust its caches again while the change it made is still nowhere. + await persistPendingBumps(); + + const table = epochTable(); + const rows = await executor() + .select({ revision: table.revision, generation: table.generation }) + .from(table) + .where(eq(table.id, RBAC_EPOCH_ROW_ID)) + .limit(1); + // A missing row is not a failure: the table exists and nothing has + // invalidated yet, which is epoch zero. + // Adopted whole, and never maxed against a local value. The row is the only + // authority: taking the larger of the two is what let a process that had + // invalidated while degraded stay permanently ahead of everyone else, and + // taking only the number is what let a REPLACED store read as the same one. + revision = rows.length > 0 ? Number(rows[0].revision) : 0; + generation = rows.length > 0 ? String(rows[0].generation) : ""; + readAt = Date.now(); + degraded = false; + } catch (error) { + reportDegraded(error); + // Rate-limit the FAILING path too. Left unset, a missing table means one + // failing query per authorization check rather than one per interval, + // which is the upgrade window turned into a load problem. + readAt = Date.now(); + } + return currentEpoch(); +} + +/** + * Push the invalidations this process owes into the shared row. + * + * ## Exactly one drain runs at a time, and this depends on it + * + * The count is read, sent, and subtracted around an await. Two drains + * overlapping would each subtract a backlog the other had already sent, taking + * the count NEGATIVE: nothing could be served from cache again, and the next + * invalidation would publish an increment of zero, so the change that raised it + * would reach no other instance at all. + * + * What rules that out is not a lock here but the shape of the only path in. + * This is called from one place, the top of {@link readShared}; `readShared` is + * called from one place, {@link refreshEpoch}; and that call sits between a + * check of `inFlight` and its assignment with no await in between, so a second + * read cannot start while one is running. A second caller added anywhere else + * would need its own answer to this, and the subtraction is where it would go + * wrong. + * + * ## The count stays owed until the row has it + * + * Subtracted only after the statement resolves, so a write nobody accepted + * leaves this process distrusting its caches rather than believing it has + * caught up — and so does a write still in flight, which the row has not + * accepted yet either. + */ +async function persistPendingBumps(): Promise { + const owed = pendingBumps; + if (owed === 0) return; + + const table = epochTable(); + // ONE statement, so there is no row count to read and no create-or-update + // branch to get wrong. The previous shape asked the driver how many rows an + // UPDATE touched and inserted when the answer was zero, which is three + // different result shapes across three drivers and a silent no-op whenever + // one of them is misread — the counter then sticks at its first value and + // every later invalidation is lost, while every individual statement + // succeeds. An upsert cannot have that failure: the row is created if it is + // absent and incremented if it is present, decided by the database. + const insert = executor().insert(table).values({ + id: RBAC_EPOCH_ROW_ID, + revision: owed, + // Only ever written when the row is CREATED; the conflict branch below + // leaves it alone, so a live counter keeps its identity for life. + generation: randomUUID(), + updatedAt: new Date(), + }); + const raise = { + target: table.id, + set: { + revision: sql`${table.revision} + ${owed}`, + updatedAt: new Date(), + }, + }; + if (typeof insert.onConflictDoUpdate === "function") { + await insert.onConflictDoUpdate(raise); + } else if (typeof insert.onDuplicateKeyUpdate === "function") { + // MySQL spells the same statement differently and takes no target. + await insert.onDuplicateKeyUpdate({ set: raise.set }); + } else { + await insert; + } + + pendingBumps -= owed; +} + +/** + * Record that authorization data changed, install-wide. + * + * The increment is one statement the database evaluates itself, so two + * instances invalidating at the same moment produce two increments rather than + * one lost update — which a read-modify-write from here would not. + * + * The value this process then answers with is READ BACK rather than assumed. + * Inventing `epoch + 1` locally is what made a degraded instance diverge, and + * it is also simply wrong whenever somebody else bumped in the same interval. + * Spelled as a FORCED refresh rather than as a second write path, because + * pushing the backlog and adopting what the row then says are already the two + * halves of one read: {@link refreshEpoch} drains before it queries, and + * forcing is what guarantees the query it makes is one this change is inside + * rather than an older one already in flight. + * + * A failed write leaves the count owed rather than applied, and + * {@link epochIsTrustworthy} then reports that nothing here may be served from + * cache until the shared row accepts it. + */ +export async function bumpEpoch(): Promise { + pendingBumps += 1; + return refreshEpoch({ force: true }); +} + +/** + * Forget everything this process knows, for a test that needs a clean one. + * + * The module holds the epoch, the time it was read and whether the table has + * been found unreadable, and a suite asserting any of those has to be able to + * start from nothing. Exported rather than reached through a mock because the + * state is this module's own. + */ +export function resetEpochForTests(): void { + revision = 0; + generation = ""; + readAt = 0; + degraded = false; + pendingBumps = 0; + inFlight = null; +} diff --git a/packages/nextly/src/services/lib/super-admin-cache-invalidation.integration.test.ts b/packages/nextly/src/services/lib/super-admin-cache-invalidation.integration.test.ts index eee825bdcf..8882fb17ea 100644 --- a/packages/nextly/src/services/lib/super-admin-cache-invalidation.integration.test.ts +++ b/packages/nextly/src/services/lib/super-admin-cache-invalidation.integration.test.ts @@ -18,12 +18,13 @@ */ import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import { createTestNextly, type TestNextly } from "nextly/testing"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { getDialectTables } from "../../database/index"; import { ApiKeyService } from "../../domains/auth/services/api-key-service"; +import { PermissionCacheService } from "../../domains/auth/services/permission-cache-service"; import { PermissionService } from "../../domains/auth/services/permission-service"; import { @@ -32,7 +33,14 @@ import { invalidatePermissionCache, isSuperAdmin, rbacRevision, + resolvedUnderCurrentRevision, } from "./permissions"; +import { + EPOCH_TTL_MS, + currentEpoch, + refreshEpoch, + resetEpochForTests, +} from "./rbac-epoch"; let harness: TestNextly | undefined; @@ -92,6 +100,10 @@ async function demote(userId: string): Promise { } beforeEach(async () => { + // The database is rebuilt per test and the epoch module is not, so its cached + // stamp would outlive the counter it describes — and a fresh counter can read + // identically to the one before it. Reset together or the two disagree. + resetEpochForTests(); harness = await createTestNextly(); await seedSuperAdminRole(); }); @@ -142,13 +154,22 @@ describe("the super-admin answer is invalidated with the permissions it is asked expect(await isSuperAdmin(userId)).toBe(false); }); - it("leaves an unrelated user's answer alone on a userId invalidation", async () => { - // The discriminating control: an implementation that clears the whole map - // for every hint passes the two cases above and is a different behaviour. - const kept = "super-cache-kept"; - const other = "super-cache-other"; - await promote(kept, "super-cache-kept@example.com"); - await promote(other, "super-cache-other@example.com"); + it("retires every in-memory answer, not only the user it names", async () => { + // A deliberate loss of scoping, and the reason is structural rather than a + // shortcut: the counter other instances read carries a number and not a + // user id, so a change they see cannot be narrower than "something in RBAC + // moved". Keeping the scope locally while broadcasting a global signal + // would mean the instance that made the change was the only one applying it + // narrowly, which is the inconsistency this replaced. + // + // The cost is a role change emptying the in-memory tiers rather than one + // entry. It is bounded by how often roles change, which is rarely, and by + // what refilling costs, which is a couple of indexed queries. A stale + // answer costs the install a grant it revoked. + const kept = `scope-kept-${randomUUID()}`; + const other = `scope-other-${randomUUID()}`; + await promote(kept, `${kept}@example.com`); + await promote(other, `${other}@example.com`); expect(await isSuperAdmin(kept)).toBe(true); expect(await isSuperAdmin(other)).toBe(true); @@ -156,8 +177,8 @@ describe("the super-admin answer is invalidated with the permissions it is asked await demote(other); await invalidatePermissionCache({ userId: other }); - expect(await isSuperAdmin(kept), "still cached").toBe(true); - expect(await isSuperAdmin(other), "evicted").toBe(false); + expect(await isSuperAdmin(other), "the named user").toBe(false); + expect(await isSuperAdmin(kept), "and everyone else").toBe(false); }); }); @@ -369,6 +390,11 @@ describe("an API key's grants are retired when the roles behind them change", () // resolution, and whether the next read sees it says whether that entry // was served or re-resolved. await seedDeputy(); + // Warm the stamp first. Capturing it is a database read now, so an unwarmed + // resolution can still be waiting on that read when the invalidation lands + // and would then capture the value AFTER it — which is a race in the test's + // setup rather than the behaviour under test. + await refreshEpoch(); const inFlight = grants(); await invalidateAllPermissionCaches(); await inFlight; @@ -469,18 +495,48 @@ describe("an API key's grants are retired when the roles behind them change", () * previous one had already expired. */ describe("a sweep of permission writes", () => { - it("advances the revision per write, so nothing in flight files as current", async () => { - // Deferring the table write must NOT defer the revision: a resolution - // running alongside the batch has to be refused, and the counter is what - // refuses it. + it("files nothing as current while the batch is open", async () => { + // A resolution running alongside the batch has to be refused. What refuses + // it is not the counter: the counter stays where it is until the stored + // rows have actually been retired, because announcing sooner tells every + // other instance to drop its own answer and read one of those rows. harness = harness ?? (await createTestNextly()); const before = rbacRevision(); + const cacheable: boolean[] = []; + const announced: string[] = []; + await inPermissionSweep(async () => { await invalidateAllPermissionCaches(); + cacheable.push(resolvedUnderCurrentRevision(rbacRevision())); + announced.push(rbacRevision()); await invalidateAllPermissionCaches(); + cacheable.push(resolvedUnderCurrentRevision(rbacRevision())); + announced.push(rbacRevision()); + }); + + // Nothing was cacheable at any point inside the batch... + expect(cacheable).toEqual([false, false]); + // ...and nothing was announced while the stored rows were still live. + expect(new Set(announced)).toEqual(new Set([before])); + // The announcement happens once, on the way out, after the retirement. + expect(rbacRevision()).not.toBe(before); + }); + + it("still empties the tiers it holds in memory, per write", async () => { + // The half a batch never deferred. Emptying memory costs nothing, so an + // answer this process is already holding must not outlive the row it came + // from just because the expensive half is being batched. + const userId = `sweep-memory-${randomUUID()}`; + await promote(userId, `${userId}@example.com`); + expect(await isSuperAdmin(userId)).toBe(true); + await demote(userId); + + const insideBatch = await inPermissionSweep(async () => { await invalidateAllPermissionCaches(); + return isSuperAdmin(userId); }); - expect(rbacRevision()).toBeGreaterThan(before + 2); + + expect(insideBatch).toBe(false); }); it("clears the process caches by the time the batch returns", async () => { @@ -533,31 +589,66 @@ describe("a sweep of permission writes", () => { expect(duringBatch).toBe(false); }); - it("still defers the batch's OWN writes, which is what the batch is for", async () => { - // The control. Without it the case above is satisfied by a sweep that - // defers nothing at all, which would pass it while removing the batching - // this exists to provide. - const userId = `inside-${randomUUID()}`; - await promote(userId, `${userId}@example.com`); - expect(await isSuperAdmin(userId)).toBe(true); - await demote(userId); + it("still defers the expensive table write, which is what the batch is for", async () => { + // The control, on the deferral that remains. The in-memory tiers no longer + // wait for the batch — the epoch moves the moment anything invalidates, so + // a demoted user stops reading as an admin immediately, inside a batch or + // out of it. What a batch still saves is the unfiltered rewrite of every + // stored row, which is the cost it was built for. + // + // Observed on a row put there directly, because `setCachedPermission` does + // not take effect under `createTestNextly`. A far-future expiry is what the + // flush overwrites, so its survival IS the deferral. + const db = harness!.adapter.getDrizzle() as unknown as { + insert: (t: unknown) => { values: (row: unknown) => Promise }; + select: (p: unknown) => { + from: (t: unknown) => { where: (c: unknown) => Promise }; + }; + }; + const tables = getDialectTables(); + const rowId = `defer-${randomUUID()}`; + const owner = `defer-user-${randomUUID()}`; + await promote(owner, `${owner}@example.com`); + const farFuture = new Date(Date.now() + 3_600_000); + await db.insert(tables.userPermissionCache).values({ + id: rowId, + userId: owner, + action: "read", + resource: "notes", + hasPermission: true, + roleIds: "[]", + expiresAt: farFuture, + createdAt: new Date(), + }); + + const stillFuture = async () => { + const rows = (await db + .select({ expiresAt: tables.userPermissionCache.expiresAt }) + .from(tables.userPermissionCache) + .where(eq(tables.userPermissionCache.id, rowId))) as Array<{ + expiresAt: Date | number; + }>; + const value = rows[0]?.expiresAt; + const ms = value instanceof Date ? value.getTime() : Number(value) * 1000; + return ms > Date.now() + 60_000; + }; let releaseBatch: () => void = () => {}; const batchRunning = new Promise(resolve => { releaseBatch = resolve; }); const batch = inPermissionSweep(async () => { - // This one IS the batch's own, so it waits for the batch to end. await invalidateAllPermissionCaches(); - const answeredInside = await isSuperAdmin(userId); + const duringBatch = await stillFuture(); await batchRunning; - return answeredInside; + return duringBatch; }); releaseBatch(); - // The batch's own write cleared nothing while the batch was open, so the - // demoted user still reads as a super admin from cache inside it. - expect(await batch).toBe(true); + // Untouched while the batch was open... + expect(await batch, "deferred during the batch").toBe(true); + // ...and rewritten on the way out, so deferred is not skipped. + expect(await stillFuture(), "flushed on exit").toBe(false); }); it("flushes even when the batch throws, since a partial write still changed rows", async () => { @@ -576,3 +667,260 @@ describe("a sweep of permission writes", () => { expect(await isSuperAdmin(userId)).toBe(false); }); }); + +/** + * The stored tier is emptied BEFORE the epoch that retires the rest is + * published. + * + * The epoch is the only signal another instance receives, and it is a one-way + * barrier: an answer filed before it moves is retired by the move, an answer + * filed after it is not. Announce first and every other instance gets a window + * in which it rejects its own in-memory answer, reads the stored row the + * tombstone has not reached yet, and files that retired decision under the NEW + * epoch, where nothing still to happen can reach it. Empty the stored rows + * first and both halves close: an instance that has not seen the move files + * under the old epoch and the move retires it, and one that has seen the move + * finds nothing to file. + * + * The interleaving is the subject, and both steps have finished by the time the + * call returns, so it cannot be seen from outside. The real retirement is left + * to run and only asked which epoch it ran under, which is the one fact that + * tells the two orders apart. + */ +describe("the order the two tiers are retired in", () => { + type Tombstone = (...args: never[]) => Promise; + + async function epochsAround( + method: "invalidateAll" | "invalidateByUser", + invalidate: () => Promise + ): Promise<{ before: string; during: string; after: string }> { + const proto = PermissionCacheService.prototype as unknown as Record< + string, + Tombstone + >; + const original = proto[method]; + const before = rbacRevision(); + let during = ""; + proto[method] = function (this: unknown, ...args: never[]) { + during = rbacRevision(); + return original.apply(this, args); + }; + try { + await invalidate(); + } finally { + proto[method] = original; + } + return { before, during, after: rbacRevision() }; + } + + it("empties the stored rows before it announces the new epoch", async () => { + const { before, during, after } = await epochsAround("invalidateAll", () => + invalidateAllPermissionCaches() + ); + + // The tombstone ran while the epoch was still the one every instance had. + expect(during).toBe(before); + // And the announcement did happen, so the assertion above is not passing + // because nothing moved at all. + expect(after).not.toBe(before); + }); + + it("raises when the stored tier refuses, rather than reporting nothing done", async () => { + // The half the cases below cannot reach. They replace `invalidateAll` with + // a rejection to exercise the caller, which proves nothing about whether + // the real one rejects — and it used to catch its own database error and + // answer `0`, which is also what a successful tombstone over an empty table + // answers. The caller could not tell those apart, so it published either + // way. This asks the real method, over an adapter whose write fails. + const real = harness!.adapter; + const unwritable = Object.create(real) as typeof real; + (unwritable as { getDrizzle: unknown }).getDrizzle = () => ({ + update: () => ({ + set: () => Promise.reject(new Error("the stored tier is unreachable")), + }), + }); + + const service = new PermissionCacheService(unwritable, console, { + cacheTtlSeconds: 300, + }); + + await expect(service.invalidateAll()).rejects.toThrow( + "the stored tier is unreachable" + ); + }); + + it("answers a count when the write succeeds, so raising is not its only mode", async () => { + // The control. Without it, "rejects on failure" is equally satisfied by a + // method that rejects on every call. + const service = new PermissionCacheService(harness!.adapter, console, { + cacheTtlSeconds: 300, + }); + + await expect(service.invalidateAll()).resolves.toBeTypeOf("number"); + }); + + it("announces nothing when the stored rows could not be retired", async () => { + // The epoch means "everything filed before this is gone". A tombstone that + // failed leaves those rows live, so the sentence is false — and announcing + // it anyway is worse than silence: every other instance rejects its own + // in-memory answer BECAUSE the epoch moved, reads one of those rows, and + // files it under the new epoch where nothing left to happen can reach it. + const proto = PermissionCacheService.prototype as unknown as Record< + string, + Tombstone + >; + const original = proto.invalidateAll; + const before = rbacRevision(); + proto.invalidateAll = () => + Promise.reject(new Error("the stored tier is unreachable")); + try { + await invalidateAllPermissionCaches(); + } finally { + proto.invalidateAll = original; + } + + expect(rbacRevision()).toBe(before); + }); + + it("announces when it CAN retire them, which is what makes silence a choice", async () => { + // The control. Without it, "announces nothing on failure" is equally + // satisfied by an epoch that never moves at all. + const before = rbacRevision(); + + await invalidateAllPermissionCaches(); + + expect(rbacRevision()).not.toBe(before); + }); + + it("still empties its own memory when the stored tier refuses", async () => { + // Withholding the announcement must not withhold the local retirement: + // this process knows the rows changed whatever the shared tier did. + const userId = `unreachable-${randomUUID()}`; + await promote(userId, `${userId}@example.com`); + expect(await isSuperAdmin(userId)).toBe(true); + await demote(userId); + + const proto = PermissionCacheService.prototype as unknown as Record< + string, + Tombstone + >; + const original = proto.invalidateAll; + proto.invalidateAll = () => + Promise.reject(new Error("the stored tier is unreachable")); + try { + await invalidateAllPermissionCaches(); + } finally { + proto.invalidateAll = original; + } + + expect(await isSuperAdmin(userId)).toBe(false); + }); + + it("does the same for an invalidation scoped to one user", async () => { + // A scoped tombstone is smaller, not faster: it is still an awaited write, + // so announcing ahead of it opens the same window. + const userId = `order-${randomUUID()}`; + await promote(userId, `${userId}@example.com`); + + const { before, during, after } = await epochsAround( + "invalidateByUser", + () => invalidatePermissionCache({ userId }) + ); + + expect(during).toBe(before); + expect(after).not.toBe(before); + }); +}); + +/** + * The counter is shared, so a change made ELSEWHERE retires what is cached here. + * + * This is the property the module left memory for, and it cannot be observed + * from one process by ordinary means: everything a test calls bumps the local + * copy as a side effect. So the other instance is played by writing the shared + * row directly — which is exactly what a second process's bump looks like from + * this one — and the only thing that can then retire the cached answer is a + * refresh reading a value this process never set. + */ +describe("an epoch bumped by another instance", () => { + /** Advance the shared counter without touching this process's copy. */ + async function bumpElsewhere(): Promise { + const db = harness!.adapter.getDrizzle() as unknown as { + update: (table: unknown) => { + set: (patch: unknown) => { where: (cond: unknown) => Promise }; + }; + insert: (table: unknown) => { + values: (row: unknown) => Promise; + }; + }; + const tables = getDialectTables() as unknown as { + nextlyRbacEpoch: { id: unknown; revision: unknown; generation: unknown }; + }; + const table = tables.nextlyRbacEpoch; + // The number alone is not the stamp: the row's generation is half of it, so + // a bump made here has to move the number and leave the identity alone, + // exactly as another instance's bump would. + try { + await db.insert(table).values({ + id: "global", + revision: 1, + generation: `another-instance-${randomUUID()}`, + updatedAt: new Date(), + }); + } catch { + // A row already exists, which is the ordinary case once anything has + // invalidated here. Move its number without touching its identity. + await db + .update(table) + .set({ revision: sql`${table.revision} + 5`, updatedAt: new Date() }) + .where(eq(table.id as never, "global")); + } + } + + it("is visible here once the read interval has passed", async () => { + // The subject. Nothing in this process bumped anything, so a local counter + // would answer with what it last set and never move. + const before = currentEpoch(); + await bumpElsewhere(); + + await new Promise(resolve => setTimeout(resolve, EPOCH_TTL_MS + 50)); + const after = await refreshEpoch(); + + expect(after).not.toBe(before); + }); + + it("retires a super-admin answer this process had cached", async () => { + // What the counter is FOR, end to end: the demotion happens by a raw row + // delete, so nothing in this process clears the cache, and the answer flips + // only because the shared counter moved. + const userId = `elsewhere-${randomUUID()}`; + await seedSuperAdminRole().catch(() => {}); + await promote(userId, `${userId}@example.com`); + expect(await isSuperAdmin(userId)).toBe(true); + + await demote(userId); + // Still cached: nothing has told this process anything changed. + expect(await isSuperAdmin(userId)).toBe(true); + + await bumpElsewhere(); + await new Promise(resolve => setTimeout(resolve, EPOCH_TTL_MS + 50)); + + expect(await isSuperAdmin(userId)).toBe(false); + }); + + it("keeps serving inside the interval, which is what makes the read cheap", async () => { + // The control on both cases above. Without it, "the answer changed" is + // equally satisfied by reading the shared row on every single check — the + // per-request query this design exists to avoid — and the interval would be + // free to regress to zero unnoticed. + const userId = `within-${randomUUID()}`; + await promote(userId, `${userId}@example.com`); + expect(await isSuperAdmin(userId)).toBe(true); + + await demote(userId); + await bumpElsewhere(); + + // No wait. The shared row has moved and this process has not looked. + expect(await isSuperAdmin(userId)).toBe(true); + }); +});