fix(libsql): serialize factory writes on shared clients

Route Factory transactions and autocommit operations through the existing per-client write queue. This prevents concurrent dispatcher claims and other shared-client writes from colliding with open LibSQL transactions or being rolled back with unrelated work.

Add file-backed regression coverage for concurrent Factory transactions and writes racing a rollback.

Co-Authored-By: Mastra Code (openai/gpt-5.6-sol) <noreply@mastra.ai>

Author: actions-user

.changeset/rich-rats-admire.md

@@ -0,0 +1,5 @@
+---
+'@mastra/libsql': patch
+---
+
+Fixed Factory database lock errors by serializing writes that share a LibSQL connection.

stores/libsql/src/storage/factory-storage.test.ts

@@ -1,8 +1,162 @@
+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
 import { describeFactoryStorageContract } from '@internal/storage-test-utils';
+import type { CollectionSchema } from '@mastra/core/storage';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
 
 import { LibSQLFactoryStorage } from './factory-storage';
 
 describeFactoryStorageContract('libsql', async () => {
   const storage = new LibSQLFactoryStorage({ url: ':memory:' });
   return { storage, close: () => storage.close() };
 });
+
+const recordsSchema = {
+  name: 'factory_write_lock_records',
+  columns: {
+    id: { type: 'uuid-pk' },
+    value: { type: 'text' },
+  },
+} satisfies CollectionSchema;
+
+type TestRecord = {
+  id: string;
+  value: string;
+};
+
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>(done => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+describe('LibSQLFactoryStorage shared-client write lock', () => {
+  let tmpDir: string;
+  let storage: LibSQLFactoryStorage;
+
+  beforeEach(async () => {
+    tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'libsql-factory-write-lock-'));
+    storage = new LibSQLFactoryStorage({
+      id: 'factory-write-lock-regression',
+      url: `file:${path.join(tmpDir, 'factory.db')}`,
+    });
+    await storage.ensureCollections([recordsSchema]);
+  });
+
+  afterEach(async () => {
+    await storage.close();
+    fs.rmSync(tmpDir, { recursive: true, force: true });
+  });
+
+  it('serializes concurrent Factory transactions', async () => {
+    const firstEntered = deferred();
+    const releaseFirst = deferred();
+
+    const first = storage.withTransaction(async ops => {
+      await ops.insertOne<TestRecord>(recordsSchema.name, { value: 'first' });
+      firstEntered.resolve();
+      await releaseFirst.promise;
+    });
+    await firstEntered.promise;
+
+    let secondEntered = false;
+    const second = storage.withTransaction(async ops => {
+      secondEntered = true;
+      await ops.insertOne<TestRecord>(recordsSchema.name, { value: 'second' });
+    });
+    void second.catch(() => {});
+
+    await Promise.resolve();
+    expect(secondEntered).toBe(false);
+
+    releaseFirst.resolve();
+    await Promise.all([first, second]);
+
+    const rows = await storage.ops.findMany<TestRecord>(recordsSchema.name, {});
+    expect(rows.map(row => row.value).sort()).toEqual(['first', 'second']);
+  });
+
+  it('keeps a Factory autocommit write outside a rolled-back Factory transaction', async () => {
+    const transactionEntered = deferred();
+    const releaseTransaction = deferred();
+    const rollback = new Error('roll back Factory transaction');
+
+    const transaction = storage.withTransaction(async ops => {
+      await ops.insertOne<TestRecord>(recordsSchema.name, { value: 'rolled-back' });
+      transactionEntered.resolve();
+      await releaseTransaction.promise;
+      throw rollback;
+    });
+    void transaction.catch(() => {});
+    await transactionEntered.promise;
+
+    let autocommitSettled = false;
+    const autocommit = storage.ops.insertOne<TestRecord>(recordsSchema.name, { value: 'kept' }).then(row => {
+      autocommitSettled = true;
+      return row;
+    });
+    void autocommit.catch(() => {});
+
+    await Promise.resolve();
+    expect(autocommitSettled).toBe(false);
+
+    releaseTransaction.resolve();
+    await expect(transaction).rejects.toBe(rollback);
+    await expect(autocommit).resolves.toMatchObject({ value: 'kept' });
+
+    await expect(storage.ops.findOne<TestRecord>(recordsSchema.name, { value: 'rolled-back' })).resolves.toBeNull();
+    await expect(storage.ops.findOne<TestRecord>(recordsSchema.name, { value: 'kept' })).resolves.toMatchObject({
+      value: 'kept',
+    });
+  });
+
+  it('keeps a wrapped Mastra-domain write outside a rolled-back Factory transaction', async () => {
+    const experiments = storage.getMastraStorage().stores?.experiments;
+    if (!experiments) throw new Error('LibSQLStore experiments domain is unavailable');
+    await experiments.init();
+
+    const transactionEntered = deferred();
+    const releaseTransaction = deferred();
+    const rollback = new Error('roll back Factory transaction');
+
+    const transaction = storage.withTransaction(async ops => {
+      await ops.insertOne<TestRecord>(recordsSchema.name, { value: 'rolled-back' });
+      transactionEntered.resolve();
+      await releaseTransaction.promise;
+      throw rollback;
+    });
+    void transaction.catch(() => {});
+    await transactionEntered.promise;
+
+    let experimentSettled = false;
+    const experiment = experiments
+      .createExperiment({
+        datasetId: 'factory-write-lock-dataset',
+        datasetVersion: 1,
+        targetType: 'agent',
+        targetId: 'factory-write-lock-agent',
+        totalItems: 1,
+      })
+      .then(record => {
+        experimentSettled = true;
+        return record;
+      });
+    void experiment.catch(() => {});
+
+    await Promise.resolve();
+    expect(experimentSettled).toBe(false);
+
+    releaseTransaction.resolve();
+    await expect(transaction).rejects.toBe(rollback);
+    const created = await experiment;
+
+    const listed = await experiments.listExperiments({
+      datasetId: 'factory-write-lock-dataset',
+      pagination: { page: 0, perPage: 10 },
+    });
+    expect(listed.experiments.map(record => record.id)).toContain(created.id);
+  });
+});

stores/libsql/src/storage/factory-storage.ts

@@ -16,6 +16,7 @@ import type {
 } from '@mastra/core/storage';
 
 import { DEFAULT_CONNECTION_TIMEOUT_MS } from './db';
+import { withClientWriteLock } from './db/write-lock';
 import { LibSQLStore } from './index';
 
 export interface LibSQLFactoryStorageConfig {
@@ -99,27 +100,20 @@ function serializeDefault(value: string | number | boolean): string {
   return String(value);
 }
 
-/** Simple FIFO in-process mutex serializing the single-writer paths. */
-class Mutex {
-  #tail: Promise<unknown> = Promise.resolve();
-
-  run<T>(fn: () => Promise<T>): Promise<T> {
-    const result = this.#tail.then(fn, fn);
-    this.#tail = result.catch(() => {});
-    return result;
-  }
-}
-
 type LibSQLExecutor = Pick<Client, 'execute'>;
+type WriteGate = <T>(fn: () => Promise<T>) => Promise<T>;
+
+const runWithoutWriteLock: WriteGate = fn => fn();
 
 class LibSQLFactoryStorageOps implements FactoryStorageOps {
   readonly #client: LibSQLExecutor;
   readonly #schemas: Map<string, CollectionSchema>;
-  readonly #writeMutex = new Mutex();
+  readonly #withWriteLock: WriteGate;
 
-  constructor(client: LibSQLExecutor, schemas: Map<string, CollectionSchema>) {
+  constructor(client: LibSQLExecutor, schemas: Map<string, CollectionSchema>, withWriteLock: WriteGate) {
     this.#client = client;
     this.#schemas = schemas;
+    this.#withWriteLock = withWriteLock;
   }
 
   #schema(collection: string): CollectionSchema {
@@ -291,7 +285,7 @@ class LibSQLFactoryStorageOps implements FactoryStorageOps {
     return this.#select<T>(collection, where, opts);
   }
 
-  async insertOne<T extends Record<string, unknown>>(collection: string, row: Partial<T>): Promise<T> {
+  async #insertOne<T extends Record<string, unknown>>(collection: string, row: Partial<T>): Promise<T> {
     const schema = this.#schema(collection);
     const pk = primaryKeyOf(schema);
 
@@ -320,7 +314,11 @@ class LibSQLFactoryStorageOps implements FactoryStorageOps {
     return inserted;
   }
 
-  async upsertOne<T extends Record<string, unknown>>(
+  async insertOne<T extends Record<string, unknown>>(collection: string, row: Partial<T>): Promise<T> {
+    return this.#withWriteLock(() => this.#insertOne<T>(collection, row));
+  }
+
+  async #upsertOne<T extends Record<string, unknown>>(
     collection: string,
     conflictKeys: string[],
     row: Partial<T>,
@@ -340,14 +338,14 @@ class LibSQLFactoryStorageOps implements FactoryStorageOps {
           ),
         );
         if (Object.keys(set).length > 0) {
-          await this.updateMany(collection, { [pk]: existing[pk] as CollectionValue }, set);
+          await this.#updateMany(collection, { [pk]: existing[pk] as CollectionValue }, set);
         }
         const updated = await this.findOne<T>(collection, { [pk]: existing[pk] as CollectionValue });
         if (!updated) continue; // deleted concurrently; retry
         return updated;
       }
       try {
-        return await this.insertOne<T>(collection, row);
+        return await this.#insertOne<T>(collection, row);
       } catch (error) {
         if (!(error instanceof UniqueViolationError)) throw error;
         lastError = error; // lost an insert race; retry as update
@@ -356,7 +354,15 @@ class LibSQLFactoryStorageOps implements FactoryStorageOps {
     throw lastError ?? new Error(`LibSQLFactoryStorage: upsert into '${collection}' did not converge`);
   }
 
-  async updateMany(collection: string, where: CollectionWhere, set: Record<string, unknown>): Promise<number> {
+  async upsertOne<T extends Record<string, unknown>>(
+    collection: string,
+    conflictKeys: string[],
+    row: Partial<T>,
+  ): Promise<T> {
+    return this.#withWriteLock(() => this.#upsertOne<T>(collection, conflictKeys, row));
+  }
+
+  async #updateMany(collection: string, where: CollectionWhere, set: Record<string, unknown>): Promise<number> {
     const schema = this.#schema(collection);
     const columns = Object.keys(set).filter(column => set[column] !== undefined);
     if (columns.length === 0) return 0;
@@ -367,7 +373,11 @@ class LibSQLFactoryStorageOps implements FactoryStorageOps {
     return result.rowsAffected;
   }
 
-  async deleteMany(collection: string, where: CollectionWhere): Promise<number> {
+  async updateMany(collection: string, where: CollectionWhere, set: Record<string, unknown>): Promise<number> {
+    return this.#withWriteLock(() => this.#updateMany(collection, where, set));
+  }
+
+  async #deleteMany(collection: string, where: CollectionWhere): Promise<number> {
     const schema = this.#schema(collection);
     const filter = this.#buildWhere(schema, where);
     const result = await this.#client.execute({
@@ -377,22 +387,24 @@ class LibSQLFactoryStorageOps implements FactoryStorageOps {
     return result.rowsAffected;
   }
 
+  async deleteMany(collection: string, where: CollectionWhere): Promise<number> {
+    return this.#withWriteLock(() => this.#deleteMany(collection, where));
+  }
+
   async updateAtomic<T extends Record<string, unknown>>(
     collection: string,
     where: CollectionWhere,
     fn: (row: T) => Partial<T> | null | Promise<Partial<T> | null>,
   ): Promise<T | null> {
-    const schema = this.#schema(collection);
-    const pk = primaryKeyOf(schema);
-    // libsql local is single-writer; serializing read-modify-write in process
-    // gives the same "no lost updates" guarantee pg gets from FOR UPDATE.
-    return this.#writeMutex.run(async () => {
+    return this.#withWriteLock(async () => {
+      const schema = this.#schema(collection);
+      const pk = primaryKeyOf(schema);
       const row = await this.findOne<T>(collection, where);
       if (!row) return null;
       const patch = await fn(row);
       if (patch === null) return row;
       const pkWhere = { [pk]: row[pk] as CollectionValue } as CollectionWhere;
-      await this.updateMany(collection, pkWhere, patch);
+      await this.#updateMany(collection, pkWhere, patch);
       return this.findOne<T>(collection, pkWhere);
     });
   }
@@ -420,7 +432,7 @@ export class LibSQLFactoryStorage extends FactoryStorage {
       ...(config.authToken ? { authToken: config.authToken } : {}),
       ...(isLocalDb ? { timeout: DEFAULT_CONNECTION_TIMEOUT_MS } : {}),
     });
-    this.ops = new LibSQLFactoryStorageOps(this.#client, this.#schemas);
+    this.ops = new LibSQLFactoryStorageOps(this.#client, this.#schemas, fn => withClientWriteLock(this.#client, fn));
   }
 
   getMastraStorage(): MastraCompositeStore {
@@ -437,25 +449,31 @@ export class LibSQLFactoryStorage extends FactoryStorage {
   }
 
   async withTransaction<T>(fn: (ops: FactoryStorageOps) => Promise<T>): Promise<T> {
-    if (this.#config.url.includes(':memory:')) return fn(this.ops);
-    const transaction = await this.#client.transaction('write');
-    try {
-      const result = await fn(new LibSQLFactoryStorageOps(transaction, this.#schemas));
-      await transaction.commit();
-      return result;
-    } catch (error) {
-      await transaction.rollback();
-      throw error;
-    } finally {
-      transaction.close();
-    }
+    return withClientWriteLock(this.#client, async () => {
+      if (this.#config.url.includes(':memory:')) {
+        return fn(new LibSQLFactoryStorageOps(this.#client, this.#schemas, runWithoutWriteLock));
+      }
+      const transaction = await this.#client.transaction('write');
+      try {
+        const result = await fn(new LibSQLFactoryStorageOps(transaction, this.#schemas, runWithoutWriteLock));
+        await transaction.commit();
+        return result;
+      } catch (error) {
+        await transaction.rollback();
+        throw error;
+      } finally {
+        transaction.close();
+      }
+    });
   }
 
   async ensureCollections(schemas: CollectionSchema[]): Promise<void> {
-    for (const schema of schemas) {
-      await this.#ensureCollection(schema);
-      this.#schemas.set(schema.name, schema);
-    }
+    await withClientWriteLock(this.#client, async () => {
+      for (const schema of schemas) {
+        await this.#ensureCollection(schema);
+        this.#schemas.set(schema.name, schema);
+      }
+    });
   }
 
   async close(): Promise<void> {