fix(tll): atomic append with advisory lock + unique position index (review #1)

Replace the read-then-insert pattern in TllRepository.append() with an
atomic INSERT...SELECT serialized by pg_advisory_xact_lock keyed on
(user_id, entity_id). Concurrent appends to the same entity chain now
serialize on the lock and compute predecessor + position from the latest
committed row, eliminating the TOCTOU race where two parallel callers
read the same MAX(position_in_chain) and both wrote at tip+1.

Add a UNIQUE (user_id, entity_id, position_in_chain) index as
defense-in-depth so any future code path that bypasses the lock fails
loudly at the DB layer rather than silently producing duplicate
positions.

Add an integration test that fires three concurrent appends to the same
chain via Promise.all and asserts positions 0,1,2 with correctly wired
predecessor pointers.

Author: moralespanitz

src/db/__tests__/repository-tll.test.ts

@@ -124,6 +124,27 @@ describe('TllRepository', () => {
 
       expect(await countRows(ent.id, mem.id)).toBe(1);
     });
+
+    it('serializes concurrent appends to the same chain without duplicate positions', async () => {
+      const ent = await makeEntity('Concurrent', 121);
+      const mA = await makeMemory('concA', 122, new Date('2026-05-01'));
+      const mB = await makeMemory('concB', 123, new Date('2026-05-02'));
+      const mC = await makeMemory('concC', 124, new Date('2026-05-03'));
+
+      await Promise.all([
+        tllRepo.append(TEST_USER, mA.id, [ent.id], mA.date),
+        tllRepo.append(TEST_USER, mB.id, [ent.id], mB.date),
+        tllRepo.append(TEST_USER, mC.id, [ent.id], mC.date),
+      ]);
+
+      const events = await tllRepo.chain(TEST_USER, ent.id);
+      expect(events).toHaveLength(3);
+      expect(events.map((e) => e.positionInChain)).toEqual([0, 1, 2]);
+      expect(new Set(events.map((e) => e.memoryId)).size).toBe(3);
+      expect(events[0].predecessorMemoryId).toBeNull();
+      expect(events[1].predecessorMemoryId).toBe(events[0].memoryId);
+      expect(events[2].predecessorMemoryId).toBe(events[1].memoryId);
+    });
   });
 
   describe('chain()', () => {

src/db/repository-tll.ts

@@ -6,9 +6,6 @@
  * paying the full graph-DB cost. Each new memory referencing an entity
  * appends an event node; the predecessor pointer lets us traverse the
  * chain backward at query time for EO/MSR/TR questions.
- *
- * The CROME proposal calls this primitive necessary for temporal reasoning
- * at 10M scale; Mem0 explicitly admits their architecture lacks it.
  */
 
 import pg from 'pg';
@@ -20,13 +17,25 @@ export interface TLLEvent {
   positionInChain: number;
 }
 
+// Stable namespace for pg_advisory_xact_lock keying. Keeps TLL appends from
+// colliding with unrelated advisory-lock callers in the same process.
+const TLL_ADVISORY_LOCK_NAMESPACE = 0x544c4c00; // "TLL\0"
+
 export class TllRepository {
   constructor(private pool: pg.Pool) {}
 
   /**
    * Append an event node to each entity's chain. Idempotent on
    * (user_id, entity_id, memory_id). Predecessor is the most-recent
    * existing event for the entity; position is len(chain).
+   *
+   * Race-safety: each (user_id, entity_id) append runs inside a transaction
+   * guarded by `pg_advisory_xact_lock`. Concurrent appends targeting the
+   * same chain serialize on the lock, then compute the next position from
+   * committed rows via INSERT...SELECT. The
+   * `(user_id, entity_id, position_in_chain)` unique index is the
+   * defense-in-depth backstop that fails loudly if any caller bypasses the
+   * lock path.
    */
   async append(
     userId: string,
@@ -37,47 +46,59 @@ export class TllRepository {
     if (entityIds.length === 0) return;
     const uniqueEntities = [...new Set(entityIds)];
 
-    // Find current chain tip per entity in one query
-    const tipResult = await this.pool.query(
-      `SELECT entity_id,
-              memory_id AS predecessor_memory_id,
-              position_in_chain
-       FROM temporal_linkage_list t1
-       WHERE user_id = $1 AND entity_id = ANY($2::uuid[])
-         AND position_in_chain = (
-           SELECT MAX(position_in_chain)
-           FROM temporal_linkage_list t2
-           WHERE t2.user_id = t1.user_id AND t2.entity_id = t1.entity_id
-         )`,
-      [userId, uniqueEntities],
-    );
-    const tips = new Map<string, { predecessorId: string; position: number }>();
-    for (const row of tipResult.rows) {
-      tips.set(row.entity_id, {
-        predecessorId: row.predecessor_memory_id,
-        position: row.position_in_chain,
-      });
+    // Per-entity transactions keep the advisory-lock scope narrow and let
+    // independent chains proceed in parallel.
+    for (const entityId of uniqueEntities) {
+      await this.appendOne(userId, memoryId, entityId, observationDate);
     }
+  }
 
-    // Batch insert
-    const values: string[] = [];
-    const params: unknown[] = [];
-    let p = 1;
-    for (const entityId of uniqueEntities) {
-      const tip = tips.get(entityId);
-      const predecessor = tip ? tip.predecessorId : null;
-      const position = tip ? tip.position + 1 : 0;
-      values.push(`($${p}, $${p + 1}, $${p + 2}, $${p + 3}, $${p + 4}, $${p + 5})`);
-      params.push(userId, entityId, memoryId, predecessor, observationDate, position);
-      p += 6;
+  /**
+   * Append one (entity, memory) row under an advisory lock keyed on the
+   * chain. The INSERT...SELECT computes predecessor + position inline from
+   * the latest committed row, so the read-then-write window the previous
+   * implementation exposed cannot reorder concurrent appends.
+   */
+  private async appendOne(
+    userId: string,
+    memoryId: string,
+    entityId: string,
+    observationDate: Date,
+  ): Promise<void> {
+    const client = await this.pool.connect();
+    try {
+      await client.query('BEGIN');
+      await client.query('SELECT pg_advisory_xact_lock($1, hashtext($2))', [
+        TLL_ADVISORY_LOCK_NAMESPACE,
+        `${userId}:${entityId}`,
+      ]);
+      await client.query(
+        `INSERT INTO temporal_linkage_list
+           (user_id, entity_id, memory_id, predecessor_memory_id,
+            observation_date, position_in_chain)
+         SELECT
+           $1, $2, $3,
+           (SELECT memory_id FROM temporal_linkage_list
+              WHERE user_id = $1 AND entity_id = $2
+              ORDER BY position_in_chain DESC LIMIT 1),
+           $4,
+           COALESCE(
+             (SELECT MAX(position_in_chain) FROM temporal_linkage_list
+                WHERE user_id = $1 AND entity_id = $2),
+             -1
+           ) + 1
+         ON CONFLICT (user_id, entity_id, memory_id) DO NOTHING`,
+        [userId, entityId, memoryId, observationDate],
+      );
+      await client.query('COMMIT');
+    } catch (err) {
+      await client.query('ROLLBACK').catch((rollbackErr) =>
+        console.error('[tll] rollback failed:', rollbackErr instanceof Error ? rollbackErr.message : rollbackErr),
+      );
+      throw err;
+    } finally {
+      client.release();
     }
-    await this.pool.query(
-      `INSERT INTO temporal_linkage_list
-        (user_id, entity_id, memory_id, predecessor_memory_id, observation_date, position_in_chain)
-        VALUES ${values.join(', ')}
-        ON CONFLICT (user_id, entity_id, memory_id) DO NOTHING`,
-      params,
-    );
   }
 
   /**

src/db/schema.sql

@@ -346,6 +346,13 @@ CREATE INDEX IF NOT EXISTS idx_tll_entity_chain
 CREATE INDEX IF NOT EXISTS idx_tll_memory
   ON temporal_linkage_list (memory_id);
 
+-- Defense-in-depth: unique (chain, position) so any future code path that
+-- bypasses the advisory-lock append fails at the DB layer instead of
+-- silently producing duplicate positions. Idempotent for fresh and
+-- existing schemas.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_tll_chain_position_unique
+  ON temporal_linkage_list (user_id, entity_id, position_in_chain);
+
 -- =====================================================================
 -- First-mention events (chronological topic-introduction list)
 -- =====================================================================