diff --git a/apps/server/src/database/execute-tx-after-commit.spec.ts b/apps/server/src/database/execute-tx-after-commit.spec.ts new file mode 100644 index 00000000..bb86d445 --- /dev/null +++ b/apps/server/src/database/execute-tx-after-commit.spec.ts @@ -0,0 +1,91 @@ +import { executeTx, registerAfterCommit } from './utils'; +import { KyselyDB, KyselyTransaction } from './types/kysely.types'; + +// Post-commit hook contract (#495 item 13): a side effect registered via +// registerAfterCommit must run ONLY AFTER the owning transaction commits, and a +// hook registered against a passed-through existingTrx must fire at the OUTER +// commit boundary — never inside the inner call. We fake the Kysely transaction +// runner so the ordering is observable without a real DB. + +/** + * A minimal db whose `.transaction().execute(cb)` records the commit ORDER: it + * runs `cb(trx)`, pushes 'commit' onto `log` (simulating the real commit that + * happens after the callback resolves), then returns the callback's result. + */ +function fakeDb(log: string[]): { db: KyselyDB; trx: KyselyTransaction } { + const trx = { __fakeTrx: true } as unknown as KyselyTransaction; + const db = { + transaction: () => ({ + execute: async (cb: (t: KyselyTransaction) => Promise) => { + const result = await cb(trx); + log.push('commit'); + return result; + }, + }), + } as unknown as KyselyDB; + return { db, trx }; +} + +describe('executeTx post-commit hooks', () => { + it('runs an afterCommit hook only AFTER the transaction commits', async () => { + const log: string[] = []; + const { db } = fakeDb(log); + + await executeTx(db, async (trx) => { + log.push('body'); + registerAfterCommit(trx, () => { + log.push('hook'); + }); + // The hook must NOT have run yet — the tx is still open. + expect(log).toEqual(['body']); + }); + + // Order proves post-commit: body → commit → hook (never body → hook → commit). + expect(log).toEqual(['body', 'commit', 'hook']); + }); + + it('drains hooks registered against a passed-through existingTrx at the OUTER commit', async () => { + const log: string[] = []; + const { db, trx: outerTrx } = fakeDb(log); + + await executeTx(db, async (outer) => { + // Nested executeTx reuses the outer trx: it must NOT commit or drain now. + await executeTx( + db, + async (inner) => { + registerAfterCommit(inner, () => { + log.push('inner-hook'); + }); + }, + outer, + ); + // Still inside the outer tx — the inner hook has not fired. + expect(log).toEqual([]); + }); + + // The single (outer) commit drains the hook registered on the shared trx. + expect(log).toEqual(['commit', 'inner-hook']); + // Sanity: the trx the hooks were registered against is the outer one. + expect(outerTrx).toBeDefined(); + }); + + it('a hook failure does not reject the already-committed executeTx', async () => { + const log: string[] = []; + const { db } = fakeDb(log); + + await expect( + executeTx(db, async (trx) => { + registerAfterCommit(trx, () => { + throw new Error('cache del blew up'); + }); + registerAfterCommit(trx, () => { + log.push('second-hook-still-runs'); + }); + return 'ok'; + }), + ).resolves.toBe('ok'); + + // The throwing hook is swallowed; a later hook still runs. + expect(log).toEqual(['commit', 'second-hook-still-runs']); + }); +}); diff --git a/apps/server/src/database/repos/workspace/workspace.repo.ts b/apps/server/src/database/repos/workspace/workspace.repo.ts index 89996766..b6d0fde7 100644 --- a/apps/server/src/database/repos/workspace/workspace.repo.ts +++ b/apps/server/src/database/repos/workspace/workspace.repo.ts @@ -3,7 +3,7 @@ import { CACHE_MANAGER } from '@nestjs/cache-manager'; import { Cache } from 'cache-manager'; import { InjectKysely } from 'nestjs-kysely'; import { KyselyDB, KyselyTransaction } from '../../types/kysely.types'; -import { dbOrTx } from '../../utils'; +import { dbOrTx, registerAfterCommit } from '../../utils'; import { InsertableWorkspace, UpdatableWorkspace, @@ -80,16 +80,30 @@ export class WorkspaceRepo { */ private async bustWorkspaceCache( workspace?: Pick | undefined, + trx?: KyselyTransaction, ): Promise { - try { - await this.cacheManager.del(CacheKey.WORKSPACE_SELF_HOSTED); - if (workspace?.hostname) { - await this.cacheManager.del( - CacheKey.WORKSPACE_BY_HOST(workspace.hostname), - ); + const del = async () => { + try { + await this.cacheManager.del(CacheKey.WORKSPACE_SELF_HOSTED); + if (workspace?.hostname) { + await this.cacheManager.del( + CacheKey.WORKSPACE_BY_HOST(workspace.hostname), + ); + } + } catch { + // cache is best-effort; TTL is the backstop } - } catch { - // cache is best-effort; TTL is the backstop + }; + if (trx) { + // Inside a caller transaction the write is NOT yet committed: busting now + // opens a repopulation window (a concurrent reader reloads the cache with + // the pre-commit / stale row, which then survives until TTL). Defer the del + // to the transaction's commit (drained by the owning executeTx) (#495). + registerAfterCommit(trx, del); + } else { + // No transaction: the mutation above already auto-committed, so this del is + // already post-commit. + await del(); } } @@ -180,7 +194,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -195,7 +209,7 @@ export class WorkspaceRepo { .returning(this.baseFields) .executeTakeFirst(); // Bust the cached "not found" so a fresh install / new tenant is seen at once. - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -249,7 +263,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -271,7 +285,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -326,7 +340,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -354,7 +368,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -376,7 +390,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } @@ -398,7 +412,7 @@ export class WorkspaceRepo { .where('id', '=', workspaceId) .returning(this.baseFields) .executeTakeFirst(); - await this.bustWorkspaceCache(workspace); + await this.bustWorkspaceCache(workspace, trx); return workspace; } diff --git a/apps/server/src/database/utils.ts b/apps/server/src/database/utils.ts index 84c3693b..9cc6f846 100644 --- a/apps/server/src/database/utils.ts +++ b/apps/server/src/database/utils.ts @@ -6,16 +6,76 @@ import { KyselyDB, KyselyTransaction } from './types/kysely.types'; * If an existing transaction is provided, it directly executes the callback with it. * Otherwise, it starts a new transaction using the provided database instance and executes the callback within that transaction. */ +/** + * Post-commit side-effect hooks, keyed by the transaction they were registered + * against. A WeakMap so an abandoned/never-drained transaction's entry is GC'd + * with the trx object (no leak). Used by {@link registerAfterCommit} / + * {@link executeTx}. + */ +const afterCommitHooks = new WeakMap< + KyselyTransaction, + Array<() => Promise | void> +>(); + +/** + * Register a side effect to run ONLY AFTER the transaction that owns `trx` + * commits. THE fix for "bust the cache inside the open transaction" bugs: a + * cache-invalidation (or any read-your-write-visible side effect) done while the + * writing transaction is still open opens a window where a concurrent reader + * repopulates the cache with the PRE-COMMIT (stale) row, so after commit the + * cache holds the old value until its TTL. Deferring the effect to post-commit + * closes that window. + * + * The hook is drained by the OUTERMOST {@link executeTx} that actually owns + * (created) this transaction — so registering against a passed-through + * `existingTrx` still fires at the real commit boundary, not at the inner call. + * NOTE: a hook registered against a transaction that was NOT created via + * `executeTx` (untracked) will never be drained — always create transactions + * through `executeTx` when you rely on post-commit hooks. + */ +export function registerAfterCommit( + trx: KyselyTransaction, + hook: () => Promise | void, +): void { + const existing = afterCommitHooks.get(trx); + if (existing) existing.push(hook); + else afterCommitHooks.set(trx, [hook]); +} + export async function executeTx( db: KyselyDB, callback: (trx: KyselyTransaction) => Promise, existingTrx?: KyselyTransaction, ): Promise { if (existingTrx) { - return await callback(existingTrx); // Execute callback with existing transaction - } else { - return await db.transaction().execute((trx) => callback(trx)); // Start new transaction and execute callback + // Reuse the caller's transaction. Any post-commit hooks registered here are + // drained by the OUTER executeTx that created `existingTrx`, at the true + // commit boundary — so we must NOT drain them now. + return await callback(existingTrx); } + // We OWN this transaction: run the body, then (only once it has COMMITTED) + // drain the post-commit hooks registered against it during the body. + let ownTrx: KyselyTransaction | undefined; + const result = await db.transaction().execute((trx) => { + ownTrx = trx; + return callback(trx); + }); + if (ownTrx) { + const hooks = afterCommitHooks.get(ownTrx); + if (hooks) { + afterCommitHooks.delete(ownTrx); + for (const hook of hooks) { + // Best-effort: a failed side effect (e.g. a cache del) must not fail the + // already-committed transaction. + try { + await hook(); + } catch { + // swallow — the durable write already committed + } + } + } + } + return result; } /*