fix(turso): stop stale migration locks from blocking all writes

README.md

@@ -58,8 +58,10 @@ On first startup after upgrading, opencode-mem automatically migrates existing m
 - Each shard is backed up as `<shard>.db.legacy.bak` before rewrite
 - Progress is tracked per shard in `<shard>.db.turso-migrate.json`
 - A global marker `.turso-migrated` is written only after all shards verify successfully
+- Normal inserts and deletes do not invalidate a completed shard migration; startup still checks that its vectors are readable and match its stored dimensions
 - Do not run multiple OpenCode instances against the same `storagePath` during migration; a lock file (`.turso-migrate.lock`) prevents concurrent migration
 - Manual dimension migrations use `.turso-operation.lock`; other plugin processes reject new memory writes until the migration finishes
+- Locks expire after 30 minutes without a progress timestamp update (or file modification time if the timestamp is unavailable), even if the owning process is alive. This lease uses wall-clock time: a forward clock step greater than 30 minutes can expire a progressing operation early; a backward step or future-dated lock can extend an inert lock's life. Locks carry a unique acquisition id, and stale cleanup only removes the exact file it inspected. Migrations heartbeat as they progress and abort if they discover that ownership was lost. Stale-lock cleanup is logged for diagnosis
 
 If migration is interrupted, the next startup resumes from the backup automatically.
 

src/services/memory-portability-service.ts

@@ -20,7 +20,10 @@ import { extractScopeFromContainerTag } from "./memory-scope.js";
 import { getProjectTagInfo } from "./tags.js";
 import { embeddingService } from "./embedding.js";
 import { ensureTursoReady } from "./turso/ready.js";
-import { acquireTursoOperationLock } from "./turso/operation-lock.js";
+import {
+  acquireTursoOperationLock,
+  isTursoLockOwnershipLostError,
+} from "./turso/operation-lock.js";
 import { tursoShardManager } from "./turso/shard-manager.js";
 import { tursoConnectionManager } from "./turso/connection-manager.js";
 import { tursoVectorSearch } from "./turso/vector-search.js";
@@ -334,6 +337,7 @@ export class MemoryPortabilityService {
           },
           isPinned: Boolean(memory.isPinned),
         });
+        releaseLock.heartbeat();
       }
 
       // Phase B: single transactional write. Avoid getWriteShard() here because it
@@ -355,13 +359,15 @@ export class MemoryPortabilityService {
               });
             }
             insertedIds.push(item.record.id);
+            releaseLock.heartbeat();
           }
         });
 
         await tursoShardManager.setVectorCount(
           shard.id,
           await tursoVectorSearch.countAllVectors(db)
         );
+        releaseLock.heartbeat();
       });
 
       log("Memory import completed", {
@@ -378,8 +384,11 @@ export class MemoryPortabilityService {
         containerTag: target.tag,
       };
     } catch (error) {
-      // Best-effort cleanup if a partial insert somehow escaped the transaction.
-      if (insertedIds.length > 0) {
+      // Best-effort cleanup for ordinary failures only. Ownership loss rolls back
+      // an open transaction, but a successor may already have committed these IDs
+      // (or our transaction may have committed before the final heartbeat).
+      // Never compensate shared rows without current ownership.
+      if (!isTursoLockOwnershipLostError(error) && insertedIds.length > 0) {
         try {
           const shards = await tursoShardManager.getAllShards(scope, hash);
           for (const shard of shards) {

src/services/migration-service.ts

@@ -105,13 +105,15 @@ export class MigrationService {
     }
 
     this.isRunning = true;
-    let releaseOperationLock: (() => void) | undefined;
+    let releaseOperationLock: ReturnType<typeof acquireTursoOperationLock> | undefined;
     this.progressCallback = progressCallback;
     const startTime = Date.now();
 
     try {
       releaseOperationLock = acquireTursoOperationLock(`dimension-${strategy}`);
+      const heartbeat = releaseOperationLock.heartbeat;
       const mismatch = await this.detectDimensionMismatch();
+      heartbeat();
 
       if (!mismatch.needsMigration) {
         return {
@@ -124,10 +126,10 @@ export class MigrationService {
       }
 
       if (strategy === "fresh-start") {
-        return await this.freshStartMigration(mismatch, startTime);
+        return await this.freshStartMigration(mismatch, startTime, heartbeat);
       }
 
-      return await this.reEmbedMigration(mismatch, startTime);
+      return await this.reEmbedMigration(mismatch, startTime, heartbeat);
     } catch (error) {
       log("Migration: failed", { error: String(error) });
       return {
@@ -147,7 +149,8 @@ export class MigrationService {
 
   private async freshStartMigration(
     mismatch: DimensionMismatch,
-    startTime: number
+    startTime: number,
+    heartbeat: () => void
   ): Promise<MigrationResult> {
     this.reportProgress({
       phase: "preparing",
@@ -158,6 +161,7 @@ export class MigrationService {
     let deletedShards = 0;
 
     for (const [index, shardInfo] of mismatch.shardMismatches.entries()) {
+      heartbeat();
       this.reportProgress({
         phase: "cleanup",
         processed: index,
@@ -174,6 +178,7 @@ export class MigrationService {
         archivePath,
       });
       deletedShards++;
+      heartbeat();
     }
 
     this.reportProgress({
@@ -193,9 +198,11 @@ export class MigrationService {
 
   private async reEmbedMigration(
     mismatch: DimensionMismatch,
-    startTime: number
+    startTime: number,
+    heartbeat: () => void
   ): Promise<MigrationResult> {
     await embeddingService.warmup();
+    heartbeat();
     embeddingService.clearCache();
 
     const totalMemories = mismatch.shardMismatches.reduce(
@@ -213,6 +220,7 @@ export class MigrationService {
     let processedCount = 0;
 
     for (const shardInfo of mismatch.shardMismatches) {
+      heartbeat();
       this.reportProgress({
         phase: "re-embedding",
         processed: processedCount,
@@ -229,7 +237,8 @@ export class MigrationService {
         shard.scope,
         shard.scopeHash,
         () =>
-          this.rebuildShardSafely(shard, (processed) => {
+          this.rebuildShardSafely(shard, heartbeat, (processed) => {
+            heartbeat();
             processedCount++;
             this.reportProgress({
               phase: "re-embedding",
@@ -244,6 +253,7 @@ export class MigrationService {
           })
       );
       reEmbeddedCount += migrated;
+      heartbeat();
     }
 
     this.reportProgress({
@@ -263,6 +273,7 @@ export class MigrationService {
 
   private async rebuildShardSafely(
     shard: ShardInfo,
+    heartbeat: () => void,
     onProcessed: (memoryId: string) => void
   ): Promise<number> {
     const sourceDb = await tursoConnectionManager.getConnection(shard.dbPath);
@@ -331,6 +342,7 @@ export class MigrationService {
           onProcessed(String(memory.id));
         }
 
+        heartbeat();
         await stagedDb.transaction("write", async (tx) => {
           for (const item of stagedBatch) {
             await tursoVectorSearch.insertVectorInTransaction(tx, item.record);
@@ -342,6 +354,7 @@ export class MigrationService {
             }
           }
         });
+        heartbeat();
       }
 
       const countRow = await stagedDb.get(`SELECT COUNT(*) AS count FROM memories`);
@@ -353,12 +366,14 @@ export class MigrationService {
       }
 
       await tursoConnectionManager.closeConnection(stagedPath);
+      heartbeat();
       writeFileSync(
         swapStatePath,
         JSON.stringify({ dbPath: shard.dbPath, stagedPath, backupPath }),
         "utf-8"
       );
       await tursoConnectionManager.closeConnection(shard.dbPath);
+      heartbeat();
       await withSqliteFileLockRetry(() => renameSqliteDatabase(shard.dbPath, backupPath));
       try {
         await withSqliteFileLockRetry(() => renameSqliteDatabase(stagedPath, shard.dbPath));

src/services/shard-path-migration-service.ts

@@ -4,7 +4,10 @@ import { CONFIG } from "../config.js";
 import { extractScopeFromContainerTag } from "./memory-scope.js";
 import { getProjectTagInfo } from "./tags.js";
 import { ensureTursoReady } from "./turso/ready.js";
-import { acquireTursoOperationLock } from "./turso/operation-lock.js";
+import {
+  acquireTursoOperationLock,
+  isTursoLockOwnershipLostError,
+} from "./turso/operation-lock.js";
 import { tursoShardManager } from "./turso/shard-manager.js";
 import { tursoConnectionManager } from "./turso/connection-manager.js";
 import { tursoVectorSearch } from "./turso/vector-search.js";
@@ -280,6 +283,7 @@ export class ShardPathMigrationService {
 
     try {
       // Stage complete, updated copies while originals remain untouched.
+      // A gigantic shard's synchronous copy or single UPDATE/count can still delay a heartbeat.
       for (const move of moves) {
         await tursoConnectionManager.closeConnection(move.oldPath);
         if (!existsSync(move.oldPath)) {
@@ -317,9 +321,11 @@ export class ShardPathMigrationService {
           );
         }
         await tursoConnectionManager.closeConnection(move.stagedPath);
+        releaseLock.heartbeat();
       }
 
       writeSwapState(statePath, state);
+      releaseLock.heartbeat();
 
       // Empty targets remain recoverable until the complete swap succeeds.
       for (let index = 0; index < targetShards.length; index++) {
@@ -328,17 +334,20 @@ export class ShardPathMigrationService {
           "pre-path-migrate",
           archivedTargets[index]!.archivePath!
         );
+        releaseLock.heartbeat();
       }
 
       for (const move of moves) {
         await tursoConnectionManager.closeConnection(move.oldPath);
+        releaseLock.heartbeat();
         await withSqliteFileLockRetry(() => renameSqliteDatabase(move.oldPath, move.backupPath));
         try {
           await withSqliteFileLockRetry(() => renameSqliteDatabase(move.stagedPath, move.newPath));
         } catch (error) {
           await withSqliteFileLockRetry(() => renameSqliteDatabase(move.backupPath, move.oldPath));
           throw error;
         }
+        releaseLock.heartbeat();
 
         await tursoShardManager.reassignShardScope(
           move.shardId,
@@ -354,6 +363,7 @@ export class ShardPathMigrationService {
       }
 
       unlinkSync(statePath);
+      releaseLock.heartbeat();
 
       // Prompt metadata is secondary and must not invalidate a completed,
       // recoverable memory migration.
@@ -380,12 +390,26 @@ export class ShardPathMigrationService {
         actions,
       };
     } catch (error) {
-      await this.rollbackSwap(state, statePath);
+      const ownershipLost = isTursoLockOwnershipLostError(error);
+      if (ownershipLost) {
+        // A successor may already be writing to the migrated paths. Only fresh
+        // ownership at startup may recover any journal that remains.
+        log(
+          "WARNING: Path migration interrupted after lock ownership lost; leaving recoverable state untouched",
+          { error: String(error), statePath }
+        );
+      } else {
+        await this.rollbackSwap(state, statePath);
+      }
       return {
         success: false,
         ...resultBase,
         dryRun: false,
-        error: error instanceof Error ? error.message : String(error),
+        error: ownershipLost
+          ? `Path migration interrupted after lock ownership lost and may have completed; state is recoverable. ${String(error)}`
+          : error instanceof Error
+            ? error.message
+            : String(error),
       };
     } finally {
       releaseLock();

src/services/turso/encryption-migrator.ts

@@ -73,7 +73,7 @@ async function canOpenPlain(dbPath: string): Promise<boolean> {
   }
 }
 
-async function dumpAndReloadEncrypted(dbPath: string): Promise<void> {
+async function dumpAndReloadEncrypted(dbPath: string, heartbeat: () => void): Promise<void> {
   const encryption = resolveDatabaseEncryption();
   if (!encryption) {
     throw new Error("databaseEncryptionKey is required for encryption migration");
@@ -101,6 +101,7 @@ async function dumpAndReloadEncrypted(dbPath: string): Promise<void> {
     );
     const tableNames = tables.map((row) => String(row.name));
     for (const table of tableNames) {
+      heartbeat();
       const create = await sourceDb.get<{ sql?: string }>(
         `SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?`,
         [table]
@@ -171,6 +172,7 @@ async function dumpAndReloadEncrypted(dbPath: string): Promise<void> {
               ]
             );
           }
+          heartbeat();
         }
         continue;
       }
@@ -184,6 +186,7 @@ async function dumpAndReloadEncrypted(dbPath: string): Promise<void> {
           `INSERT OR REPLACE INTO ${table} (${columns.join(", ")}) VALUES (${placeholders})`,
           columns.map((col) => asSqlValue(row[col]))
         );
+        heartbeat();
       }
     }
 
@@ -201,17 +204,21 @@ async function dumpAndReloadEncrypted(dbPath: string): Promise<void> {
       } catch {
         // may already exist
       }
+      heartbeat();
     }
   } finally {
     await sourceDb.close();
     await stagedDb.close();
   }
 
+  // Pre-swap aborts preserve the plaintext original and may leave a staged temp file.
+  heartbeat();
   await withSqliteFileLockRetry(() => {
     copySqliteDatabase(dbPath, backupPath);
     removeSqliteDatabase(dbPath);
     renameSqliteDatabase(stagedPath, dbPath);
   });
+  heartbeat();
 
   // Verify encrypted open
   const verify = await connect(dbPath, buildConnectOptions(encryption));
@@ -231,17 +238,20 @@ export async function runDatabaseEncryptionMigration(): Promise<void> {
 
   const candidates = collectCandidateDbs();
   const release = acquireTursoOperationLock("tursodb-encrypt");
+  const heartbeat = release.heartbeat;
   try {
     for (const path of candidates) {
+      heartbeat();
       if (await isAlreadyEncrypted(path)) continue;
       if (!(await canOpenPlain(path))) {
         throw new Error(
           `Cannot open ${path} as plaintext for encryption migration. ` +
             `If it is already encrypted with a different key, update databaseEncryptionKey.`
         );
       }
-      await dumpAndReloadEncrypted(path);
+      await dumpAndReloadEncrypted(path, heartbeat);
     }
+    heartbeat();
     writeFileSync(
       markerPath(),
       JSON.stringify({