fix(cache): bustWorkspaceCache — после коммита, не внутри транзакции

bustWorkspaceCache звался сразу после write, но ВНУТРИ переданной транзакции
(до коммита). Окно: параллельный читатель промахивается мимо инвалидированного
ключа, читает ещё НЕ закоммиченную (старую) строку и репопулирует кэш старым
значением; после коммита кэш держит устаревшее до TTL (15 c).

Добавил post-commit-хук в executeTx: registerAfterCommit(trx, fn) регистрирует
side-effect, который дренится ТОЛЬКО после коммита транзакции — причём внешним
executeTx, владеющим trx (проброшенный existingTrx срабатывает на настоящей
границе коммита, а не во вложенном вызове). WeakMap по trx — без утечки.
bustWorkspaceCache теперь: без trx — del сразу (уже автокоммит); с trx —
регистрирует del на post-commit. Ошибка хука не валит уже закоммиченный write.

Тест: порядок body→commit→hook, дренаж на внешней границе, глушение падения хука.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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<unknown>) => {
+        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']);
+  });
+});

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<Workspace, 'hostname'> | undefined,
+    trx?: KyselyTransaction,
   ): Promise<void> {
-    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;
   }
 

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> | 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,
+): void {
+  const existing = afterCommitHooks.get(trx);
+  if (existing) existing.push(hook);
+  else afterCommitHooks.set(trx, [hook]);
+}
+
 export async function executeTx<T>(
   db: KyselyDB,
   callback: (trx: KyselyTransaction) => Promise<T>,
   existingTrx?: KyselyTransaction,
 ): Promise<T> {
   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;
 }
 
 /*