From ca9e1ad3285a28230d6bfaa752d6fbc61e303dc3 Mon Sep 17 00:00:00 2001 From: Phototonic <35551999+Phototonic@users.noreply.github.com> Date: Mon, 26 Oct 2026 09:46:44 +1100 Subject: [PATCH 1/5] fix(turso): preserve completed migrations after normal writes Treat complete sidecars as historical import verification, not a live row-count invariant. Keep native vector readiness checks and cover backup restoration for unreadable completed shards. --- README.md | 1 + src/services/turso/legacy-migrator.ts | 8 +-- tests/turso-legacy-migrator.test.ts | 85 +++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 26e4498c..b0af3184 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,7 @@ On first startup after upgrading, opencode-mem automatically migrates existing m - Each shard is backed up as `.db.legacy.bak` before rewrite - Progress is tracked per shard in `.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 diff --git a/src/services/turso/legacy-migrator.ts b/src/services/turso/legacy-migrator.ts index b4da0747..7dbc16ef 100644 --- a/src/services/turso/legacy-migrator.ts +++ b/src/services/turso/legacy-migrator.ts @@ -210,11 +210,9 @@ async function isShardMigrationComplete(dbPath: string): Promise { if (!(await hasMemoriesTable(db))) { return sidecar.expectedCount === 0; } - const count = await countMemories(db); - if (count !== sidecar.expectedCount) { - return false; - } - // Sidecar alone is not enough — require native vector index + readable vectors. + // Counts are verified before publishing the complete sidecar. Normal + // inserts and deletes must not invalidate that historical migration fact. + // Sidecar alone is not enough — still require native, readable vectors. return isTursoVectorShardReady(db); } catch { return false; diff --git a/tests/turso-legacy-migrator.test.ts b/tests/turso-legacy-migrator.test.ts index cfcca22e..8f0dc6ce 100644 --- a/tests/turso-legacy-migrator.test.ts +++ b/tests/turso-legacy-migrator.test.ts @@ -106,6 +106,91 @@ describe("turso legacy migrator", () => { expect(registered[0]?.vectorCount).toBe(1); }); + for (const change of ["insert", "delete"] as const) { + migrationTest(`keeps migration complete after a normal ${change}`, async () => { + baseDir = mkdtempSync(join(tmpdir(), "turso-migrate-completed-")); + const projectsDir = join(baseDir, "projects"); + mkdirSync(projectsDir); + const scopeHash = "abcdef0123456789"; + const dbPath = await createLegacyShard(projectsDir, `project_${scopeHash}_shard_0.db`, [ + { + id: "original", + vector: new Float32Array([1, 0, 0, 0]), + content: "original memory", + containerTag: `opencode_project_${scopeHash}`, + }, + ]); + const { CONFIG } = await import("../src/config.js"); + CONFIG.storagePath = baseDir; + CONFIG.embeddingDimensions = 4; + CONFIG.databaseEncryptionEnabled = false; + CONFIG.databaseEncryptionKey = undefined; + const { runLegacyTursoMigration } = await import("../src/services/turso/legacy-migrator.js"); + const { tursoConnectionManager } = + await import("../src/services/turso/connection-manager.js"); + await runLegacyTursoMigration(); + const sidecarPath = `${dbPath}.turso-migrate.json`; + const sidecar = readFileSync(sidecarPath, "utf-8"); + const markerPath = join(baseDir, ".turso-migrated"); + const marker = readFileSync(markerPath, "utf-8"); + const db = await tursoConnectionManager.getConnection(dbPath); + if (change === "insert") { + await db.run(`INSERT INTO memories (id, content, vector, container_tag, created_at, updated_at) + SELECT 'added', content, vector, container_tag, created_at, updated_at FROM memories`); + } else { + await db.run("DELETE FROM memories"); + } + + // A live lock makes any erroneous re-entry fail rather than silently + // rewriting the historical counts. A completed shard needs no migration. + writeFileSync( + join(baseDir, ".turso-migrate.lock"), + JSON.stringify({ pid: process.pid, timestamp: new Date().toISOString() }) + ); + const { closeTursoAndInvalidateCaches } = await import("../src/services/turso/lifecycle.js"); + await closeTursoAndInvalidateCaches(); + await runLegacyTursoMigration(); + expect(readFileSync(sidecarPath, "utf-8")).toBe(sidecar); + expect(readFileSync(markerPath, "utf-8")).toBe(marker); + const { tursoShardManager } = await import("../src/services/turso/shard-manager.js"); + const shards = await tursoShardManager.getAllShards("project", scopeHash); + expect(shards[0]?.vectorCount).toBe(change === "insert" ? 2 : 0); + }); + } + + migrationTest("restores unreadable vectors despite a complete sidecar", async () => { + baseDir = mkdtempSync(join(tmpdir(), "turso-migrate-complete-corrupt-")); + const projectsDir = join(baseDir, "projects"); + mkdirSync(projectsDir); + const dbPath = await createLegacyShard(projectsDir, "project_abcdef0123456789_shard_0.db", [ + { + id: "recoverable", + vector: new Float32Array([1, 0, 0, 0]), + content: "preserved in backup", + containerTag: "opencode_project_abcdef0123456789", + }, + ]); + const { CONFIG } = await import("../src/config.js"); + CONFIG.storagePath = baseDir; + CONFIG.embeddingDimensions = 4; + CONFIG.databaseEncryptionEnabled = false; + CONFIG.databaseEncryptionKey = undefined; + const { runLegacyTursoMigration } = await import("../src/services/turso/legacy-migrator.js"); + const { tursoConnectionManager } = await import("../src/services/turso/connection-manager.js"); + await runLegacyTursoMigration(); + const db = await tursoConnectionManager.getConnection(dbPath); + await db.run("UPDATE memories SET vector = x'010203'"); + + await runLegacyTursoMigration(); + + const restoredDb = await tursoConnectionManager.getConnection(dbPath); + const row = await restoredDb.get( + "SELECT content, vector_extract(vector) AS vector FROM memories" + ); + expect(row?.content).toBe("preserved in backup"); + expect(JSON.parse(String(row?.vector))).toEqual([1, 0, 0, 0]); + }); + migrationTest("resumes from backup after interrupted migration", async () => { baseDir = mkdtempSync(join(tmpdir(), "turso-migrate-resume-")); const projectsDir = join(baseDir, "projects"); From f51d3e21f52681197bb1c456dd828d1cac20f764 Mon Sep 17 00:00:00 2001 From: Phototonic <35551999+Phototonic@users.noreply.github.com> Date: Mon, 26 Oct 2026 09:46:55 +1100 Subject: [PATCH 2/5] fix(turso): bound live-but-inert migration lock staleness Expire locks after thirty minutes without progress, with timestamp/mtime fallback and audited stale cleanup shared by write gating and legacy acquisition. Heartbeat legacy insert batches and shards plus dimension migration progress; abort on lost ownership instead of continuing concurrently. --- README.md | 1 + src/services/migration-service.ts | 27 ++++- src/services/turso/legacy-migrator.ts | 41 +++---- src/services/turso/operation-lock.ts | 114 ++++++++++++++++-- tests/turso-legacy-migrator.test.ts | 67 +++++++++++ tests/turso-operation-lock.test.ts | 163 ++++++++++++++++++++++++++ tests/turso-reembed-migration.test.ts | 49 +++++++- 7 files changed, 418 insertions(+), 44 deletions(-) create mode 100644 tests/turso-operation-lock.test.ts diff --git a/README.md b/README.md index b0af3184..b51d050d 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ On first startup after upgrading, opencode-mem automatically migrates existing m - 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. Legacy and dimension 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. diff --git a/src/services/migration-service.ts b/src/services/migration-service.ts index 23e21321..79e17c7c 100644 --- a/src/services/migration-service.ts +++ b/src/services/migration-service.ts @@ -105,13 +105,15 @@ export class MigrationService { } this.isRunning = true; - let releaseOperationLock: (() => void) | undefined; + let releaseOperationLock: ReturnType | 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 { 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 { 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 { 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)); diff --git a/src/services/turso/legacy-migrator.ts b/src/services/turso/legacy-migrator.ts index 7dbc16ef..54dcdd64 100644 --- a/src/services/turso/legacy-migrator.ts +++ b/src/services/turso/legacy-migrator.ts @@ -12,6 +12,7 @@ import { log } from "../logger.js"; import { tursoConnectionManager } from "./connection-manager.js"; import { tursoShardManager } from "./shard-manager.js"; import { runTursoEngineMigration } from "./engine-migrator.js"; +import { heartbeatTursoLock, readLiveTursoLock } from "./operation-lock.js"; import { withSqliteFileLockRetry, renameSqliteDatabase, @@ -91,36 +92,13 @@ function writeMarker(storagePath: string, shards: MigrationMarker["shards"]): vo writeFileSync(join(storagePath, MIGRATION_MARKER), JSON.stringify(marker, null, 2), "utf-8"); } -function isProcessAlive(pid: number): boolean { - try { - process.kill(pid, 0); - return true; - } catch { - return false; - } -} - function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } function acquireMigrationLock(storagePath: string): boolean { const lockPath = join(storagePath, MIGRATION_LOCK); - if (existsSync(lockPath)) { - try { - const lock = JSON.parse(readFileSync(lockPath, "utf-8")) as MigrationLock; - if (lock.pid && isProcessAlive(lock.pid)) { - return false; - } - unlinkSync(lockPath); - } catch { - try { - unlinkSync(lockPath); - } catch { - // stale or corrupt lock — try atomic create below - } - } - } + if (readLiveTursoLock(lockPath)) return false; const lock: MigrationLock = { pid: process.pid, timestamp: new Date().toISOString() }; try { @@ -151,10 +129,14 @@ function releaseMigrationLock(storagePath: string): void { unlinkSync(lockPath); } } catch { - unlinkSync(lockPath); + // Missing, replaced, corrupt, or held open: cleanup must not block writes. } } +function heartbeatMigrationLock(): void { + heartbeatTursoLock(join(CONFIG.storagePath, MIGRATION_LOCK)); +} + async function countMemories(db: TursoDb): Promise { const row = await db.get(`SELECT COUNT(*) as count FROM memories`); return Number(row?.count ?? 0); @@ -409,6 +391,7 @@ async function migrateMemoryShard(dbPath: string): Promise renameSqliteDatabase(dbPath, backup)); } @@ -435,6 +419,7 @@ async function migrateMemoryShard(dbPath: string): Promise { await acquireMigrationLockWithRetry(CONFIG.storagePath); try { + heartbeatMigrationLock(); const dbFiles = ["metadata.db", "user-prompts.db", "user-profiles.db", "ai-sessions.db"]; for (const file of dbFiles) { const path = join(CONFIG.storagePath, file); @@ -661,7 +650,9 @@ export async function runLegacyTursoMigration(): Promise { throw new Error("Turso legacy migration incomplete: not all shards verified"); } + heartbeatMigrationLock(); await reconcileShardRegistry(); + heartbeatMigrationLock(); writeMarker(CONFIG.storagePath, migratedShards); log("Turso legacy migration complete", { storagePath: CONFIG.storagePath }); } finally { diff --git a/src/services/turso/operation-lock.ts b/src/services/turso/operation-lock.ts index 35d8224b..a9f801d1 100644 --- a/src/services/turso/operation-lock.ts +++ b/src/services/turso/operation-lock.ts @@ -1,13 +1,29 @@ -import { existsSync, readFileSync, unlinkSync, writeFileSync } from "node:fs"; +import { + closeSync, + existsSync, + fstatSync, + openSync, + readFileSync, + statSync, + unlinkSync, + writeFileSync, + writeSync, +} from "node:fs"; import { join } from "node:path"; import { CONFIG } from "../../config.js"; +import { log } from "../logger.js"; const OPERATION_LOCK = ".turso-operation.lock"; const LEGACY_MIGRATION_LOCK = ".turso-migrate.lock"; +// Thirty minutes of progress silence is far longer than a local 50-row insert +// batch, shard swap, or individual embedding request. Long migrations heartbeat +// at those boundaries, so total migration duration is not limited by this lease. +export const TURSO_LOCK_STALE_MS = 30 * 60 * 1000; + interface LockState { pid: number; - timestamp: string; + timestamp?: string; operation?: string; } @@ -20,29 +36,98 @@ function isProcessAlive(pid: number): boolean { } } -function readLiveLock(path: string): LockState | null { +function lockAgeMs(path: string, state: LockState | null): number | null { + let timestamp = typeof state?.timestamp === "string" ? Date.parse(state.timestamp) : NaN; + if (!Number.isFinite(timestamp)) { + try { + timestamp = statSync(path).mtimeMs; + } catch { + // Without a usable timestamp or mtime, fail closed for a live PID. + } + } + return Number.isFinite(timestamp) ? Date.now() - timestamp : null; +} + +/** Shared lease check for both write gating and atomic migration acquisition. */ +export function readLiveTursoLock(path: string): LockState | null { if (!existsSync(path)) return null; + let state: LockState | null = null; + let reason = "corrupt lock"; try { - const state = JSON.parse(readFileSync(path, "utf-8")) as LockState; - if (Number.isInteger(state.pid) && state.pid > 0 && isProcessAlive(state.pid)) { - return state; + state = JSON.parse(readFileSync(path, "utf-8")) as LockState | null; + if (state && Number.isInteger(state.pid) && state.pid > 0) { + const age = lockAgeMs(path, state); + if (isProcessAlive(state.pid)) { + if (age === null || age <= TURSO_LOCK_STALE_MS) return state; + reason = "progress heartbeat expired"; + } else { + reason = "dead PID"; + } } } catch { // Corrupt locks are stale and removed below. } + const ageMs = lockAgeMs(path, state); + let removed = false; try { unlinkSync(path); + removed = true; } catch { // Already removed (race) or held open (Windows) — either way the lock is // gone or unowned; do not let cleanup failure block writes. } + const details = { + path, + pid: state?.pid ?? null, + ageMs, + operation: state?.operation, + reason, + removed, + }; + try { + log("WARNING: Invalidating stale Turso migration lock", details); + } catch { + // Audit even when the log destination is unavailable; logging must not + // turn an unowned lock into a permanent write outage. + console.warn("WARNING: Invalidating stale Turso migration lock", details); + } return null; } +/** Refresh progress only while this process still owns the lease. */ +export function heartbeatTursoLock(path: string): void { + let fd: number | undefined; + try { + // Open without creating, and update the same file we checked. If another + // process replaces the path, we must not overwrite its new lease. + fd = openSync(path, "r+"); + const previous = readFileSync(fd, "utf-8"); + const state = JSON.parse(previous) as LockState | null; + if (state?.pid !== process.pid) throw new Error("lock owner changed"); + const updated = JSON.stringify({ ...state, timestamp: new Date().toISOString() }); + // Keep any shortened tail as JSON whitespace rather than leaving a corrupt + // window between a rewrite and truncate (older locks may be pretty-printed). + const padding = Math.max(0, Buffer.byteLength(previous) - Buffer.byteLength(updated)); + const contents = Buffer.from(updated + " ".repeat(padding)); + writeSync(fd, contents, 0, contents.length, 0); + const opened = fstatSync(fd); + const current = statSync(path); + if (opened.dev !== current.dev || opened.ino !== current.ino) { + throw new Error("lock file replaced"); + } + } catch (error) { + throw new Error(`Turso migration lock ownership lost at ${path}; aborting migration`, { + cause: error, + }); + } finally { + if (fd !== undefined) closeSync(fd); + } +} + export function assertNoTursoMigrationInProgress(): void { for (const file of [OPERATION_LOCK, LEGACY_MIGRATION_LOCK]) { const path = join(CONFIG.storagePath, file); - const state = readLiveLock(path); + const state = readLiveTursoLock(path); if (state) { throw new Error( `Database migration is in progress${state.operation ? ` (${state.operation})` : ""} ` + @@ -52,7 +137,9 @@ export function assertNoTursoMigrationInProgress(): void { } } -export function acquireTursoOperationLock(operation: string): () => void { +export function acquireTursoOperationLock( + operation: string +): (() => void) & { heartbeat: () => void } { assertNoTursoMigrationInProgress(); const path = join(CONFIG.storagePath, OPERATION_LOCK); const state: LockState = { @@ -62,10 +149,13 @@ export function acquireTursoOperationLock(operation: string): () => void { }; writeFileSync(path, JSON.stringify(state), { flag: "wx" }); - return () => { - const current = readLiveLock(path); - if (current?.pid === process.pid && existsSync(path)) { - unlinkSync(path); + const release = () => { + try { + const current = JSON.parse(readFileSync(path, "utf-8")) as LockState | null; + if (current?.pid === process.pid) unlinkSync(path); + } catch { + // Missing, replaced, corrupt, or held open: cleanup must not block writes. } }; + return Object.assign(release, { heartbeat: () => heartbeatTursoLock(path) }); } diff --git a/tests/turso-legacy-migrator.test.ts b/tests/turso-legacy-migrator.test.ts index 8f0dc6ce..2d1d251a 100644 --- a/tests/turso-legacy-migrator.test.ts +++ b/tests/turso-legacy-migrator.test.ts @@ -191,6 +191,73 @@ describe("turso legacy migrator", () => { expect(JSON.parse(String(row?.vector))).toEqual([1, 0, 0, 0]); }); + for (const loseLock of [false, true]) { + migrationTest(`heartbeats insert batches (ownership lost: ${loseLock})`, async () => { + baseDir = mkdtempSync(join(tmpdir(), "turso-migrate-heartbeat-")); + const projectsDir = join(baseDir, "projects"); + mkdirSync(projectsDir); + const dbPath = await createLegacyShard( + projectsDir, + "project_abcdef0123456789_shard_0.db", + Array.from({ length: 51 }, (_, i) => ({ + id: `batch-${i}`, + vector: new Float32Array([1, 0, 0, 0]), + content: "batch memory", + containerTag: "opencode_project_abcdef0123456789", + })) + ); + const { CONFIG } = await import("../src/config.js"); + CONFIG.storagePath = baseDir; + CONFIG.embeddingDimensions = 4; + CONFIG.databaseEncryptionEnabled = false; + CONFIG.databaseEncryptionKey = undefined; + const { TURSO_LOCK_STALE_MS } = await import("../src/services/turso/operation-lock.js"); + const { tursoVectorSearch } = await import("../src/services/turso/vector-search.js"); + const insert = tursoVectorSearch.insertVectorInTransaction; + const lockPath = join(baseDir, ".turso-migrate.lock"); + const oldTimestamp = new Date(Date.now() - TURSO_LOCK_STALE_MS - 60_000).toISOString(); + let inserted = 0; + tursoVectorSearch.insertVectorInTransaction = async function (tx, record) { + if (inserted === 0) { + writeFileSync( + lockPath, + JSON.stringify({ + pid: loseLock ? 2147483647 : process.pid, + timestamp: oldTimestamp, + }) + ); + } + if (inserted === 50) { + const lock = JSON.parse(readFileSync(lockPath, "utf-8")); + expect(Date.parse(lock.timestamp)).toBeGreaterThan(Date.parse(oldTimestamp)); + } + inserted++; + return insert.call(this, tx, record); + }; + try { + const { runLegacyTursoMigration } = + await import("../src/services/turso/legacy-migrator.js"); + if (loseLock) { + await expect(runLegacyTursoMigration()).rejects.toThrow( + /ownership lost.*aborting migration/ + ); + expect(existsSync(join(baseDir, ".turso-migrated"))).toBe(false); + expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + } else { + await runLegacyTursoMigration(); + expect(inserted).toBe(51); + } + const { tursoConnectionManager } = + await import("../src/services/turso/connection-manager.js"); + const db = await tursoConnectionManager.getConnection(dbPath); + const row = await db.get("SELECT COUNT(*) AS count FROM memories"); + expect(Number(row?.count)).toBe(loseLock ? 0 : 51); + } finally { + tursoVectorSearch.insertVectorInTransaction = insert; + } + }); + } + migrationTest("resumes from backup after interrupted migration", async () => { baseDir = mkdtempSync(join(tmpdir(), "turso-migrate-resume-")); const projectsDir = join(baseDir, "projects"); diff --git a/tests/turso-operation-lock.test.ts b/tests/turso-operation-lock.test.ts new file mode 100644 index 00000000..acdb8456 --- /dev/null +++ b/tests/turso-operation-lock.test.ts @@ -0,0 +1,163 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, utimesSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { cleanupTursoTestDirectory } from "./turso-test-utils.js"; + +describe("turso operation lock staleness", () => { + let baseDir: string; + let restoreConfig: (() => void) | undefined; + let staleMs: number; + + afterEach(async () => { + await cleanupTursoTestDirectory(baseDir); + restoreConfig?.(); + }); + + async function withStorage() { + baseDir = mkdtempSync(join(tmpdir(), "turso-lock-staleness-")); + const { CONFIG } = await import("../src/config.js"); + const previous = { storagePath: CONFIG.storagePath }; + restoreConfig = () => Object.assign(CONFIG, previous); + CONFIG.storagePath = baseDir; + const locks = await import("../src/services/turso/operation-lock.js"); + staleMs = locks.TURSO_LOCK_STALE_MS; + return locks; + } + + const oldTimestamp = () => new Date(Date.now() - staleMs - 60_000).toISOString(); + + for (const file of [".turso-operation.lock", ".turso-migrate.lock"]) { + it(`cleans an aged-out live PID in ${file}`, async () => { + const { assertNoTursoMigrationInProgress } = await withStorage(); + const path = join(baseDir, file); + writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp: oldTimestamp() })); + expect(() => assertNoTursoMigrationInProgress()).not.toThrow(); + expect(existsSync(path)).toBe(false); + }); + + it(`blocks a fresh live PID in ${file}`, async () => { + const { assertNoTursoMigrationInProgress } = await withStorage(); + const path = join(baseDir, file); + writeFileSync( + path, + JSON.stringify({ pid: process.pid, timestamp: new Date().toISOString() }) + ); + expect(() => assertNoTursoMigrationInProgress()).toThrow(/writes are temporarily blocked/); + expect(existsSync(path)).toBe(true); + }); + + it(`cleans a dead PID and corrupt JSON in ${file}`, async () => { + const { assertNoTursoMigrationInProgress } = await withStorage(); + const path = join(baseDir, file); + // Outside supported OS PID ranges, so this cannot accidentally name a live process. + writeFileSync(path, JSON.stringify({ pid: 2147483647, timestamp: new Date().toISOString() })); + expect(() => assertNoTursoMigrationInProgress()).not.toThrow(); + expect(existsSync(path)).toBe(false); + writeFileSync(path, "{broken"); + expect(() => assertNoTursoMigrationInProgress()).not.toThrow(); + expect(existsSync(path)).toBe(false); + }); + + for (const timestamp of [undefined, "invalid"]) { + it(`uses mtime for ${file} with timestamp ${String(timestamp)}`, async () => { + const { assertNoTursoMigrationInProgress } = await withStorage(); + const path = join(baseDir, file); + writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp })); + expect(() => assertNoTursoMigrationInProgress()).toThrow(/writes are temporarily blocked/); + const old = new Date(oldTimestamp()); + utimesSync(path, old, old); + expect(() => assertNoTursoMigrationInProgress()).not.toThrow(); + expect(existsSync(path)).toBe(false); + }); + } + } + + it("allows legacy migration to acquire an aged-out live lock", async () => { + await withStorage(); + const path = join(baseDir, ".turso-migrate.lock"); + writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp: oldTimestamp() })); + const { runLegacyTursoMigration } = await import("../src/services/turso/legacy-migrator.js"); + await runLegacyTursoMigration(); + expect(existsSync(path)).toBe(false); + expect(existsSync(join(baseDir, ".turso-migrated"))).toBe(true); + }); + + it("keeps an old-acquired operation live after a progress heartbeat", async () => { + const { acquireTursoOperationLock, assertNoTursoMigrationInProgress } = await withStorage(); + const release = acquireTursoOperationLock("test-heartbeat"); + const path = join(baseDir, ".turso-operation.lock"); + const state = JSON.parse(readFileSync(path, "utf-8")); + writeFileSync(path, JSON.stringify({ ...state, timestamp: oldTimestamp() })); + release.heartbeat(); + expect(() => assertNoTursoMigrationInProgress()).toThrow(/test-heartbeat/); + expect(JSON.parse(readFileSync(path, "utf-8")).operation).toBe("test-heartbeat"); + release(); + expect(existsSync(path)).toBe(false); + }); + + it("refreshes legacy progress without relying on the acquisition timestamp", async () => { + const { heartbeatTursoLock, assertNoTursoMigrationInProgress } = await withStorage(); + const path = join(baseDir, ".turso-migrate.lock"); + writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp: oldTimestamp() }, null, 2)); + heartbeatTursoLock(path); + expect(() => assertNoTursoMigrationInProgress()).toThrow(/writes are temporarily blocked/); + }); + + it("aborts a heartbeat when the lock is missing, corrupt, or no longer ours", async () => { + const { heartbeatTursoLock } = await withStorage(); + const path = join(baseDir, ".turso-migrate.lock"); + expect(() => heartbeatTursoLock(path)).toThrow(/ownership lost.*aborting migration/); + expect(existsSync(path)).toBe(false); + for (const contents of [ + "{broken", + JSON.stringify({ pid: 2147483647, timestamp: oldTimestamp() }), + ]) { + writeFileSync(path, contents); + expect(() => heartbeatTursoLock(path)).toThrow(/ownership lost.*aborting migration/); + expect(readFileSync(path, "utf-8")).toBe(contents); + } + }); + + it("does not release another owner's operation lock", async () => { + const { acquireTursoOperationLock } = await withStorage(); + const release = acquireTursoOperationLock("test-release"); + const path = join(baseDir, ".turso-operation.lock"); + const contents = JSON.stringify({ pid: 2147483647, timestamp: new Date().toISOString() }); + writeFileSync(path, contents); + release(); + expect(readFileSync(path, "utf-8")).toBe(contents); + }); + + it("logs the expired owner, age, operation, and path for audit", async () => { + const { assertNoTursoMigrationInProgress } = await withStorage(); + const previousLogFile = process.env.OPENCODE_MEM_LOG_FILE; + const logPath = join(baseDir, "audit.log"); + process.env.OPENCODE_MEM_LOG_FILE = logPath; + try { + const path = join(baseDir, ".turso-operation.lock"); + writeFileSync( + path, + JSON.stringify({ + pid: process.pid, + timestamp: oldTimestamp(), + operation: "test-audit", + }) + ); + assertNoTursoMigrationInProgress(); + const line = readFileSync(logPath, "utf-8") + .split("\n") + .find((entry) => entry.includes("WARNING:")); + expect(line).toBeDefined(); + const details = JSON.parse(line!.slice(line!.indexOf(": {") + 2)); + expect(details.pid).toBe(process.pid); + expect(details.ageMs).toBeGreaterThan(staleMs); + expect(details.operation).toBe("test-audit"); + expect(details.path).toBe(path); + expect(details.removed).toBe(true); + } finally { + if (previousLogFile === undefined) delete process.env.OPENCODE_MEM_LOG_FILE; + else process.env.OPENCODE_MEM_LOG_FILE = previousLogFile; + } + }); +}); diff --git a/tests/turso-reembed-migration.test.ts b/tests/turso-reembed-migration.test.ts index b8a6a30c..2b4ea7b0 100644 --- a/tests/turso-reembed-migration.test.ts +++ b/tests/turso-reembed-migration.test.ts @@ -1,5 +1,12 @@ import { afterEach, describe, expect, it } from "bun:test"; -import { copyFileSync, existsSync, mkdtempSync, readdirSync, writeFileSync } from "node:fs"; +import { + copyFileSync, + existsSync, + mkdtempSync, + readdirSync, + readFileSync, + writeFileSync, +} from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { cleanupTursoTestDirectory } from "./turso-test-utils.js"; @@ -125,6 +132,46 @@ describe("turso re-embed migration safety", () => { expect(Number(metadata?.value)).toBe(2); }); + for (const loseLock of [false, true]) { + it(`heartbeats re-embed progress (ownership lost: ${loseLock})`, async () => { + const shard = await createSourceShard(); + await stubEmbedding(); + const { CONFIG } = await import("../src/config.js"); + CONFIG.embeddingDimensions = 4; + const { TURSO_LOCK_STALE_MS } = await import("../src/services/turso/operation-lock.js"); + const oldTimestamp = new Date(Date.now() - TURSO_LOCK_STALE_MS - 60_000).toISOString(); + const lockPath = join(baseDir, ".turso-operation.lock"); + const { migrationService } = await import("../src/services/migration-service.js"); + const result = await migrationService.migrateToNewModel("re-embed", (progress) => { + if (progress.phase !== "re-embedding") return; + if (progress.processed === 1) { + const lock = JSON.parse(readFileSync(lockPath, "utf-8")); + writeFileSync( + lockPath, + JSON.stringify({ + ...lock, + pid: loseLock ? 2147483647 : process.pid, + timestamp: oldTimestamp, + }) + ); + } else if (progress.processed === 2) { + const lock = JSON.parse(readFileSync(lockPath, "utf-8")); + expect(Date.parse(lock.timestamp)).toBeGreaterThan(Date.parse(oldTimestamp)); + } + }); + expect(result.success).toBe(!loseLock); + if (loseLock) { + expect(result.error).toContain("ownership lost"); + expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + const { tursoConnectionManager } = + await import("../src/services/turso/connection-manager.js"); + const db = await tursoConnectionManager.getConnection(shard.dbPath); + const row = await db.get("SELECT vector_extract(vector) AS vector FROM memories LIMIT 1"); + expect(JSON.parse(String(row?.vector))).toHaveLength(2); + } + }); + } + it("archives source shards instead of deleting them during fresh start", async () => { const shard = await createSourceShard(); const { CONFIG } = await import("../src/config.js"); From 1ee36cf14cc34d2c8f19295b04ea13e6a0a8069a Mon Sep 17 00:00:00 2001 From: Phototonic <35551999+Phototonic@users.noreply.github.com> Date: Mon, 26 Oct 2026 10:59:49 +1100 Subject: [PATCH 3/5] fix(turso): heartbeat remaining long-running operation locks --- src/services/memory-portability-service.ts | 4 ++ src/services/shard-path-migration-service.ts | 7 +++ src/services/turso/encryption-migrator.ts | 14 ++++- src/services/turso/engine-migrator.ts | 61 +++++++++++++------ tests/memory-portability.test.ts | 49 +++++++++++++++ tests/shard-path-migrate.test.ts | 54 +++++++++++++++++ tests/turso-encryption-migrator.test.ts | 63 +++++++++++++++++++- tests/turso-engine-migrator.test.ts | 60 +++++++++++++++++++ 8 files changed, 290 insertions(+), 22 deletions(-) diff --git a/src/services/memory-portability-service.ts b/src/services/memory-portability-service.ts index 686a3093..b31ee8ba 100644 --- a/src/services/memory-portability-service.ts +++ b/src/services/memory-portability-service.ts @@ -334,6 +334,7 @@ export class MemoryPortabilityService { }, isPinned: Boolean(memory.isPinned), }); + releaseLock.heartbeat(); } // Phase B: single transactional write. Avoid getWriteShard() here because it @@ -355,6 +356,7 @@ export class MemoryPortabilityService { }); } insertedIds.push(item.record.id); + releaseLock.heartbeat(); } }); @@ -362,6 +364,7 @@ export class MemoryPortabilityService { shard.id, await tursoVectorSearch.countAllVectors(db) ); + releaseLock.heartbeat(); }); log("Memory import completed", { @@ -379,6 +382,7 @@ export class MemoryPortabilityService { }; } catch (error) { // Best-effort cleanup if a partial insert somehow escaped the transaction. + // Heartbeat failures roll back the transaction; those IDs are harmless no-ops here. if (insertedIds.length > 0) { try { const shards = await tursoShardManager.getAllShards(scope, hash); diff --git a/src/services/shard-path-migration-service.ts b/src/services/shard-path-migration-service.ts index 6cfdff94..9652bfdf 100644 --- a/src/services/shard-path-migration-service.ts +++ b/src/services/shard-path-migration-service.ts @@ -280,6 +280,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 +318,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,10 +331,12 @@ 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)); @@ -339,6 +344,7 @@ export class ShardPathMigrationService { await withSqliteFileLockRetry(() => renameSqliteDatabase(move.backupPath, move.oldPath)); throw error; } + releaseLock.heartbeat(); await tursoShardManager.reassignShardScope( move.shardId, @@ -354,6 +360,7 @@ export class ShardPathMigrationService { } unlinkSync(statePath); + releaseLock.heartbeat(); // Prompt metadata is secondary and must not invalidate a completed, // recoverable memory migration. diff --git a/src/services/turso/encryption-migrator.ts b/src/services/turso/encryption-migrator.ts index 347b499d..0b4f374a 100644 --- a/src/services/turso/encryption-migrator.ts +++ b/src/services/turso/encryption-migrator.ts @@ -73,7 +73,7 @@ async function canOpenPlain(dbPath: string): Promise { } } -async function dumpAndReloadEncrypted(dbPath: string): Promise { +async function dumpAndReloadEncrypted(dbPath: string, heartbeat: () => void): Promise { const encryption = resolveDatabaseEncryption(); if (!encryption) { throw new Error("databaseEncryptionKey is required for encryption migration"); @@ -101,6 +101,7 @@ async function dumpAndReloadEncrypted(dbPath: string): Promise { ); 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 { ] ); } + heartbeat(); } continue; } @@ -184,6 +186,7 @@ async function dumpAndReloadEncrypted(dbPath: string): Promise { `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 { } 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,8 +238,10 @@ export async function runDatabaseEncryptionMigration(): Promise { 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( @@ -240,8 +249,9 @@ export async function runDatabaseEncryptionMigration(): Promise { `If it is already encrypted with a different key, update databaseEncryptionKey.` ); } - await dumpAndReloadEncrypted(path); + await dumpAndReloadEncrypted(path, heartbeat); } + heartbeat(); writeFileSync( markerPath(), JSON.stringify({ diff --git a/src/services/turso/engine-migrator.ts b/src/services/turso/engine-migrator.ts index b9fca43f..d266657f 100644 --- a/src/services/turso/engine-migrator.ts +++ b/src/services/turso/engine-migrator.ts @@ -95,7 +95,8 @@ async function needsEngineRewrite(dbPath: string): Promise { async function copyTableViaLibsql( sourcePath: string, targetDb: TursoDb, - table: string + table: string, + heartbeat: () => void ): Promise { const client = openLibsql(sourcePath); try { @@ -141,6 +142,7 @@ async function copyTableViaLibsql( sqlArgs ); imported += 1; + heartbeat(); } return imported; } finally { @@ -148,7 +150,7 @@ async function copyTableViaLibsql( } } -async function rewriteMemoryShard(dbPath: string): Promise { +async function rewriteMemoryShard(dbPath: string, heartbeat: () => void): Promise { const client = openLibsql(dbPath); let dims = CONFIG.embeddingDimensions; let model = CONFIG.embeddingModel; @@ -178,18 +180,24 @@ async function rewriteMemoryShard(dbPath: string): Promise { const staged = await tursoConnectionManager.getConnection(stagedPath); try { await tursoShardManager.initShardDb(staged, dims, model); - const imported = await copyTableViaLibsql(dbPath, staged, "memories"); + const imported = await copyTableViaLibsql(dbPath, staged, "memories", heartbeat); const metaClient = openLibsql(dbPath); try { - const metaRows = await metaClient.execute(`SELECT key, value FROM shard_metadata`); - for (const row of metaRows.rows) { - await staged.run(`INSERT OR REPLACE INTO shard_metadata (key, value) VALUES (?, ?)`, [ - String(row.key), - String(row.value), - ]); + const metaRows = await metaClient + .execute(`SELECT key, value FROM shard_metadata`) + .catch(() => null); + for (const row of metaRows?.rows ?? []) { + try { + await staged.run(`INSERT OR REPLACE INTO shard_metadata (key, value) VALUES (?, ?)`, [ + String(row.key), + String(row.value), + ]); + } catch { + // Metadata is optional; do not swallow ownership failures below. + break; + } + heartbeat(); } - } catch { - // optional } finally { metaClient.close(); } @@ -205,11 +213,14 @@ async function rewriteMemoryShard(dbPath: string): Promise { } // Match encryption-migrator: Windows cannot rename over an existing file (EPERM). + // Pre-swap aborts preserve the original and may leave a staged temp file. + heartbeat(); await withSqliteFileLockRetry(() => { copySqliteDatabase(dbPath, backupPath); removeSqliteDatabase(dbPath); renameSqliteDatabase(stagedPath, dbPath); }); + heartbeat(); log("Migrated libSQL DiskANN shard to @tursodatabase/database", { dbPath, @@ -217,7 +228,11 @@ async function rewriteMemoryShard(dbPath: string): Promise { }); } -async function rewriteGenericDb(dbPath: string, tables: string[]): Promise { +async function rewriteGenericDb( + dbPath: string, + tables: string[], + heartbeat: () => void +): Promise { const stagedPath = `${dbPath}.tursodb-migrate-${Date.now()}.tmp`; const backupPath = `${dbPath}.pre-tursodb-${Date.now()}.bak`; @@ -230,6 +245,7 @@ async function rewriteGenericDb(dbPath: string, tables: string[]): Promise const source = openLibsql(dbPath); try { for (const table of tables) { + heartbeat(); const schema = await source.execute({ sql: `SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?`, args: [table], @@ -237,7 +253,7 @@ async function rewriteGenericDb(dbPath: string, tables: string[]): Promise const createSql = schema.rows[0]?.sql ? String(schema.rows[0].sql) : null; if (!createSql) continue; await staged.run(createSql.replace(/F32_BLOB/gi, "BLOB")); - await copyTableViaLibsql(dbPath, staged, table); + await copyTableViaLibsql(dbPath, staged, table, heartbeat); } const indexes = await source.execute( `SELECT sql FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL AND name NOT LIKE 'sqlite_%'` @@ -250,6 +266,7 @@ async function rewriteGenericDb(dbPath: string, tables: string[]): Promise } catch { // index may already exist from CREATE TABLE } + heartbeat(); } } finally { source.close(); @@ -260,11 +277,14 @@ async function rewriteGenericDb(dbPath: string, tables: string[]): Promise } // Match encryption-migrator: Windows cannot rename over an existing file (EPERM). + // Pre-swap aborts preserve the original and may leave a staged temp file. + heartbeat(); await withSqliteFileLockRetry(() => { copySqliteDatabase(dbPath, backupPath); removeSqliteDatabase(dbPath); renameSqliteDatabase(stagedPath, dbPath); }); + heartbeat(); log("Migrated auxiliary DB to @tursodatabase/database", { dbPath, backupPath }); } @@ -294,23 +314,26 @@ export async function runTursoEngineMigration(): Promise { } const release = acquireTursoOperationLock("tursodb-engine-migrate"); + const heartbeat = release.heartbeat; try { for (const path of toRewrite) { + heartbeat(); const base = path.split(/[/\\]/).pop() ?? path; if (base.endsWith("_shard_") || /_shard_\d+\.db$/.test(base) || /shard_\d+\.db$/.test(base)) { - await rewriteMemoryShard(path); + await rewriteMemoryShard(path, heartbeat); } else if (base === "metadata.db") { - await rewriteGenericDb(path, ["shards"]); + await rewriteGenericDb(path, ["shards"], heartbeat); } else if (base === "user-prompts.db") { - await rewriteGenericDb(path, ["user_prompts"]); + await rewriteGenericDb(path, ["user_prompts"], heartbeat); } else if (base === "user-profiles.db") { - await rewriteGenericDb(path, ["user_profiles", "user_profile_changelogs"]); + await rewriteGenericDb(path, ["user_profiles", "user_profile_changelogs"], heartbeat); } else if (base === "ai-sessions.db") { - await rewriteGenericDb(path, ["ai_sessions", "ai_messages"]); + await rewriteGenericDb(path, ["ai_sessions", "ai_messages"], heartbeat); } else { - await rewriteMemoryShard(path); + await rewriteMemoryShard(path, heartbeat); } } + heartbeat(); writeFileSync( markerPath(), JSON.stringify({ migratedAt: new Date().toISOString(), rewritten: toRewrite }) diff --git a/tests/memory-portability.test.ts b/tests/memory-portability.test.ts index 5d79ea14..9f917cce 100644 --- a/tests/memory-portability.test.ts +++ b/tests/memory-portability.test.ts @@ -147,6 +147,55 @@ describe("memory export/import portability", () => { expect(result.rejected?.[0]?.id).toBe("mem_portable"); }); + it("aborts on ownership loss mid-embedding without importing any records", async () => { + await stubEmbedding(); + await setupProject(); + const { CONFIG } = await import("../src/config.js"); + const { memoryPortabilityService } = + await import("../src/services/memory-portability-service.js"); + const inputPath = join(baseDir, "memories.json"); + await memoryPortabilityService.exportMemories({ + currentDirectory: projectDir, + outputPath: inputPath, + }); + const document = JSON.parse(readFileSync(inputPath, "utf-8")); + document.memories = ["import_one", "import_two", "import_three"].map((id) => ({ + ...document.memories[0], + id, + tags: [], + })); + writeFileSync(inputPath, JSON.stringify(document)); + const targetDir = join(baseDir, "target"); + mkdirSync(targetDir); + const { getProjectTagInfo } = await import("../src/services/tags.js"); + const { tursoShardManager } = await import("../src/services/turso/shard-manager.js"); + const { tursoConnectionManager } = await import("../src/services/turso/connection-manager.js"); + const { tursoVectorSearch } = await import("../src/services/turso/vector-search.js"); + const hash = getProjectTagInfo(targetDir).tag.split("_").pop()!; + const shard = await tursoShardManager.createShard("project", hash, 0); + const { embeddingService } = await import("../src/services/embedding.js"); + const lockPath = join(CONFIG.storagePath, ".turso-operation.lock"); + let embedded = 0; + embeddingService.embedWithTimeout = async () => { + if (++embedded === 2) writeFileSync(lockPath, JSON.stringify({ pid: 2147483647 })); + return new Float32Array([1, 0]); + }; + const result = await memoryPortabilityService.importMemories({ + currentDirectory: targetDir, + inputPath, + }); + expect(result.success).toBe(false); + expect(result.error).toContain("ownership lost"); + expect(embedded).toBe(2); + expect( + await tursoVectorSearch.countAllVectors( + await tursoConnectionManager.getConnection(shard.dbPath) + ) + ).toBe(0); + // Release must not unlink the replacement owner's lease. + expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + }); + it("rejects unsupported schema versions", async () => { await stubEmbedding(); await setupProject(); diff --git a/tests/shard-path-migrate.test.ts b/tests/shard-path-migrate.test.ts index 149f219e..ff7aa450 100644 --- a/tests/shard-path-migrate.test.ts +++ b/tests/shard-path-migrate.test.ts @@ -4,6 +4,7 @@ import { existsSync, mkdtempSync, mkdirSync, + readFileSync, renameSync, rmSync, writeFileSync, @@ -261,6 +262,59 @@ describe("shard path migration", () => { expect(existsSync(oldShards[0]!.dbPath)).toBe(true); }); + migrationTest( + "aborts ownership loss during staging with originals and registry unchanged", + async () => { + const getProjectTagInfo = await createProjects(); + const { tag, hash } = await seedShard(oldProjectDir, [ + { id: "one", content: "first" }, + { id: "two", content: "second", shardIndex: 1 }, + ]); + const { CONFIG } = await import("../src/config.js"); + const { tursoShardManager } = await import("../src/services/turso/shard-manager.js"); + const { tursoConnectionManager } = + await import("../src/services/turso/connection-manager.js"); + const { tursoVectorSearch } = await import("../src/services/turso/vector-search.js"); + const originals = await tursoShardManager.getAllShards("project", hash); + const update = tursoVectorSearch.updateProjectAssociation; + const lockPath = join(CONFIG.storagePath, ".turso-operation.lock"); + let staged = 0; + tursoVectorSearch.updateProjectAssociation = async function (...args) { + const result = await update.apply(this, args); + staged++; + writeFileSync(lockPath, JSON.stringify({ pid: 2147483647 })); + return result; + }; + try { + const { shardPathMigrationService } = + await import("../src/services/shard-path-migration-service.js"); + const result = await shardPathMigrationService.migrate({ + currentDirectory: newProjectDir, + fromHash: hash, + allowLinkedSource: true, + }); + expect(result.success).toBe(false); + expect(result.error).toContain("ownership lost"); + } finally { + tursoVectorSearch.updateProjectAssociation = update; + } + expect(staged).toBe(1); + expect(await tursoShardManager.getAllShards("project", hash)).toEqual(originals); + const newHash = getProjectTagInfo(newProjectDir).tag.split("_").pop()!; + expect(await tursoShardManager.getAllShards("project", newHash)).toHaveLength(0); + for (const shard of originals) { + const rows = await tursoVectorSearch.getAllMemories( + await tursoConnectionManager.getConnection(shard.dbPath) + ); + expect(rows).toHaveLength(1); + expect(rows[0]?.container_tag).toBe(tag.tag); + expect(rows[0]?.project_path).toBe(oldProjectDir); + } + expect(existsSync(join(CONFIG.storagePath, ".path-migrate-swap.json"))).toBe(false); + expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + } + ); + migrationTest("refuses to migrate a still-linked source unless explicitly allowed", async () => { await createProjects(); const { hash: oldHash } = await seedShard(oldProjectDir, [ diff --git a/tests/turso-encryption-migrator.test.ts b/tests/turso-encryption-migrator.test.ts index 55fced0b..7fcb6a0f 100644 --- a/tests/turso-encryption-migrator.test.ts +++ b/tests/turso-encryption-migrator.test.ts @@ -1,5 +1,12 @@ import { afterEach, describe, expect, it } from "bun:test"; -import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync } from "node:fs"; +import { + existsSync, + mkdirSync, + mkdtempSync, + readdirSync, + readFileSync, + writeFileSync, +} from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { connect } from "@tursodatabase/database"; @@ -136,6 +143,60 @@ describe("turso encryption migrator", () => { await plain.close(); }); + migrationTest("aborts staged row copying on ownership loss and retries safely", async () => { + await withStorage(true); + const dbPath = await createPlaintextShard("project_a1b2c3d4e5f67890_shard_0.db"); + const native = await connect(dbPath, { experimental: ["encryption"] }); + const db = new TursoDb(native); + await db.run(`INSERT INTO memories (id, content, vector, container_tag, created_at, updated_at) + SELECT 'mem_second', content, vector, container_tag, created_at, updated_at FROM memories`); + await db.close(); + const { resolveOrCreateDatabaseEncryptionKey } = + await import("../src/services/turso/encryption-key.js"); + const hexkey = resolveOrCreateDatabaseEncryptionKey()!; + const run = TursoDb.prototype.run; + const lockPath = join(baseDir, ".turso-operation.lock"); + let copied = 0; + TursoDb.prototype.run = async function (sql, args) { + const result = await run.call(this, sql, args); + if (sql.includes("INSERT OR REPLACE INTO memories")) { + copied++; + writeFileSync(lockPath, JSON.stringify({ pid: 2147483647 })); + } + return result; + }; + const { runDatabaseEncryptionMigration } = + await import("../src/services/turso/encryption-migrator.js"); + try { + await expect(runDatabaseEncryptionMigration()).rejects.toThrow(/ownership lost/); + } finally { + TursoDb.prototype.run = run; + } + expect(copied).toBe(1); + expect(existsSync(join(baseDir, ".tursodb-encrypted-v1"))).toBe(false); + expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + const plain = await connect(dbPath, { experimental: ["encryption"] }); + try { + expect(await plain.prepare("SELECT content FROM memories").all()).toEqual([ + { content: "encrypt me" }, + { content: "encrypt me" }, + ]); + } finally { + await plain.close(); + } + await runDatabaseEncryptionMigration(); + expect(existsSync(join(baseDir, ".tursodb-encrypted-v1"))).toBe(true); + const encrypted = await connect(dbPath, { + experimental: ["encryption"], + encryption: { cipher: "aes256gcm", hexkey }, + }); + try { + expect(await encrypted.prepare("SELECT id FROM memories").all()).toHaveLength(2); + } finally { + await encrypted.close(); + } + }); + migrationTest("encrypts plaintext shards and skips Turso-internal tables", async () => { await withStorage(true); const dbPath = await createPlaintextShard("project_a1b2c3d4e5f67890_shard_0.db", true); diff --git a/tests/turso-engine-migrator.test.ts b/tests/turso-engine-migrator.test.ts index 6b7910c3..274cb1fc 100644 --- a/tests/turso-engine-migrator.test.ts +++ b/tests/turso-engine-migrator.test.ts @@ -157,4 +157,64 @@ describe("turso engine migrator", () => { await reopened.close(); } }); + + migrationTest("aborts staged row copying on ownership loss and retries safely", async () => { + await withStorage(); + const dbPath = join(baseDir, "metadata.db"); + const source = createClient({ url: `file:${dbPath}` }); + try { + await source.execute("CREATE TABLE shards (id TEXT PRIMARY KEY, vector F32_BLOB(4))"); + for (const id of ["one", "two", "three"]) { + await source.execute({ + sql: "INSERT INTO shards VALUES (?, vector32(?))", + args: [id, "[1,0,0,0]"], + }); + } + await source.execute("CREATE INDEX shards_vector ON shards (libsql_vector_idx(vector))"); + } finally { + source.close(); + } + + const { TursoDb } = await import("../src/services/turso/turso-db.js"); + const run = TursoDb.prototype.run; + const lockPath = join(baseDir, ".turso-operation.lock"); + let copied = 0; + TursoDb.prototype.run = async function (sql, args) { + const result = await run.call(this, sql, args); + if (sql.startsWith("INSERT OR REPLACE INTO shards")) { + copied++; + writeFileSync(lockPath, JSON.stringify({ pid: 2147483647 })); + } + return result; + }; + const { runTursoEngineMigration } = await import("../src/services/turso/engine-migrator.js"); + try { + await expect(runTursoEngineMigration()).rejects.toThrow(/ownership lost/); + } finally { + TursoDb.prototype.run = run; + } + expect(copied).toBe(1); + expect(existsSync(join(baseDir, ".tursodb-engine-v1"))).toBe(false); + expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + const original = createClient({ url: `file:${dbPath}` }); + try { + expect((await original.execute("SELECT id FROM shards")).rows).toHaveLength(3); + expect( + String( + (await original.execute("SELECT sql FROM sqlite_master WHERE name = 'shards_vector'")) + .rows[0]?.sql + ) + ).toContain("libsql_vector_idx"); + } finally { + original.close(); + } + await runTursoEngineMigration(); + expect(existsSync(join(baseDir, ".tursodb-engine-v1"))).toBe(true); + const migrated = await connect(dbPath, { experimental: ["encryption"] }); + try { + expect(await migrated.prepare("SELECT id FROM shards").all()).toHaveLength(3); + } finally { + await migrated.close(); + } + }); }); From a6f5f5342ecadc87f638f773e2474287d0998b04 Mon Sep 17 00:00:00 2001 From: Phototonic <35551999+Phototonic@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:22:48 +1100 Subject: [PATCH 4/5] fix(turso): verify lock file and acquisition ownership --- README.md | 2 +- src/services/turso/legacy-migrator.ts | 63 +++++++------- src/services/turso/operation-lock.ts | 117 ++++++++++++++++++-------- tests/turso-legacy-migrator.test.ts | 26 +++++- tests/turso-operation-lock.test.ts | 79 +++++++++++++++-- tests/turso-reembed-migration.test.ts | 1 + 6 files changed, 214 insertions(+), 74 deletions(-) diff --git a/README.md b/README.md index b51d050d..315023ee 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ On first startup after upgrading, opencode-mem automatically migrates existing m - 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. Legacy and dimension migrations heartbeat as they progress and abort if they discover that ownership was lost. Stale-lock cleanup is logged for diagnosis +- 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. diff --git a/src/services/turso/legacy-migrator.ts b/src/services/turso/legacy-migrator.ts index 54dcdd64..91305b49 100644 --- a/src/services/turso/legacy-migrator.ts +++ b/src/services/turso/legacy-migrator.ts @@ -7,12 +7,13 @@ import { writeFileSync, } from "node:fs"; import { basename, join } from "node:path"; +import { randomUUID } from "node:crypto"; import { CONFIG } from "../../config.js"; import { log } from "../logger.js"; import { tursoConnectionManager } from "./connection-manager.js"; import { tursoShardManager } from "./shard-manager.js"; import { runTursoEngineMigration } from "./engine-migrator.js"; -import { heartbeatTursoLock, readLiveTursoLock } from "./operation-lock.js"; +import { heartbeatTursoLock, readLiveTursoLock, releaseTursoLock } from "./operation-lock.js"; import { withSqliteFileLockRetry, renameSqliteDatabase, @@ -43,6 +44,7 @@ interface MigrationMarker { } interface MigrationLock { + id: string; pid: number; timestamp: string; } @@ -96,23 +98,31 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function acquireMigrationLock(storagePath: string): boolean { +function acquireMigrationLock(storagePath: string): string | null { const lockPath = join(storagePath, MIGRATION_LOCK); - if (readLiveTursoLock(lockPath)) return false; + if (readLiveTursoLock(lockPath)) return null; - const lock: MigrationLock = { pid: process.pid, timestamp: new Date().toISOString() }; + const lock: MigrationLock = { + id: randomUUID(), + pid: process.pid, + timestamp: new Date().toISOString(), + }; try { writeFileSync(lockPath, JSON.stringify(lock), { flag: "wx" }); - return true; + return lock.id; } catch { - return false; + return null; } } -async function acquireMigrationLockWithRetry(storagePath: string, maxAttempts = 3): Promise { +async function acquireMigrationLockWithRetry( + storagePath: string, + maxAttempts = 3 +): Promise { for (let attempt = 0; attempt < maxAttempts; attempt++) { - if (acquireMigrationLock(storagePath)) { - return; + const id = acquireMigrationLock(storagePath); + if (id) { + return id; } await sleep(200 + attempt * 150); } @@ -120,21 +130,8 @@ async function acquireMigrationLockWithRetry(storagePath: string, maxAttempts = throw new Error("Turso legacy migration locked by another process"); } -function releaseMigrationLock(storagePath: string): void { - const lockPath = join(storagePath, MIGRATION_LOCK); - if (!existsSync(lockPath)) return; - try { - const lock = JSON.parse(readFileSync(lockPath, "utf-8")) as MigrationLock; - if (lock.pid === process.pid) { - unlinkSync(lockPath); - } - } catch { - // Missing, replaced, corrupt, or held open: cleanup must not block writes. - } -} - -function heartbeatMigrationLock(): void { - heartbeatTursoLock(join(CONFIG.storagePath, MIGRATION_LOCK)); +function releaseMigrationLock(storagePath: string, id: string): void { + releaseTursoLock(join(storagePath, MIGRATION_LOCK), id); } async function countMemories(db: TursoDb): Promise { @@ -294,7 +291,10 @@ function rowToRecord(row: Record): MemoryRecord | null { }; } -async function migrateMemoryShard(dbPath: string): Promise { +async function migrateMemoryShard( + dbPath: string, + heartbeatMigrationLock: () => void +): Promise { if (await isShardMigrationComplete(dbPath)) { const sidecar = readSidecar(dbPath); return sidecar; @@ -456,7 +456,8 @@ async function migrateMemoryShard(dbPath: string): Promise void ): Promise> { const dir = join(CONFIG.storagePath, dirName); const results: Array<{ path: string; sidecar: ShardMigrationSidecar }> = []; @@ -466,7 +467,7 @@ async function migrateDirectory( if (!file.endsWith(".db") || file.includes(BACKUP_SUFFIX)) continue; heartbeatMigrationLock(); const dbPath = join(dir, file); - const sidecar = await migrateMemoryShard(dbPath); + const sidecar = await migrateMemoryShard(dbPath, heartbeatMigrationLock); if (sidecar) { results.push({ path: dbPath, sidecar }); } @@ -622,7 +623,9 @@ export async function runLegacyTursoMigration(): Promise { return; } - await acquireMigrationLockWithRetry(CONFIG.storagePath); + const lockId = await acquireMigrationLockWithRetry(CONFIG.storagePath); + const heartbeatMigrationLock = () => + heartbeatTursoLock(join(CONFIG.storagePath, MIGRATION_LOCK), lockId); try { heartbeatMigrationLock(); @@ -636,7 +639,7 @@ export async function runLegacyTursoMigration(): Promise { const migratedShards: MigrationMarker["shards"] = []; for (const dirName of ["users", "projects"] as const) { - const dirResults = await migrateDirectory(dirName); + const dirResults = await migrateDirectory(dirName, heartbeatMigrationLock); for (const { path, sidecar } of dirResults) { migratedShards.push({ path, @@ -656,6 +659,6 @@ export async function runLegacyTursoMigration(): Promise { writeMarker(CONFIG.storagePath, migratedShards); log("Turso legacy migration complete", { storagePath: CONFIG.storagePath }); } finally { - releaseMigrationLock(CONFIG.storagePath); + releaseMigrationLock(CONFIG.storagePath, lockId); } } diff --git a/src/services/turso/operation-lock.ts b/src/services/turso/operation-lock.ts index a9f801d1..c1357ac6 100644 --- a/src/services/turso/operation-lock.ts +++ b/src/services/turso/operation-lock.ts @@ -1,6 +1,5 @@ import { closeSync, - existsSync, fstatSync, openSync, readFileSync, @@ -8,7 +7,9 @@ import { unlinkSync, writeFileSync, writeSync, + type Stats, } from "node:fs"; +import { randomUUID } from "node:crypto"; import { join } from "node:path"; import { CONFIG } from "../../config.js"; import { log } from "../logger.js"; @@ -22,6 +23,7 @@ const LEGACY_MIGRATION_LOCK = ".turso-migrate.lock"; export const TURSO_LOCK_STALE_MS = 30 * 60 * 1000; interface LockState { + id?: string; pid: number; timestamp?: string; operation?: string; @@ -36,27 +38,75 @@ function isProcessAlive(pid: number): boolean { } } -function lockAgeMs(path: string, state: LockState | null): number | null { +function lockAgeMs(identity: Stats, state: LockState | null): number | null { let timestamp = typeof state?.timestamp === "string" ? Date.parse(state.timestamp) : NaN; if (!Number.isFinite(timestamp)) { - try { - timestamp = statSync(path).mtimeMs; - } catch { - // Without a usable timestamp or mtime, fail closed for a live PID. - } + timestamp = identity.mtimeMs; } return Number.isFinite(timestamp) ? Date.now() - timestamp : null; } +function sameFile(a: Pick, b: Pick): boolean { + return a.dev === b.dev && a.ino === b.ino; +} + +/** Remove only the file inspected by the caller; cleanup must never throw. */ +export function unlinkTursoLockIfUnchanged( + path: string, + identity: Pick +): boolean { + try { + if (!sameFile(identity, statSync(path))) return false; + // A nanosecond stat → unlink race remains without interprocess locking. + // The true owner bounds it by aborting at its next progress heartbeat. + unlinkSync(path); + return true; + } catch { + // Missing or held open (Windows): preserve best-effort cleanup semantics. + return false; + } +} + +/** Release by acquisition identity, never merely by PID. */ +export function releaseTursoLock(path: string, expectedId: string): void { + let fd: number | undefined; + try { + fd = openSync(path, "r"); + const identity = fstatSync(fd); + const state = JSON.parse(readFileSync(fd, "utf-8")) as LockState | null; + if (state?.id === expectedId) unlinkTursoLockIfUnchanged(path, identity); + } catch { + // Missing, replaced, corrupt, or held open: cleanup must not block writes. + } finally { + if (fd !== undefined) closeSync(fd); + } +} + +export class TursoLockOwnershipLostError extends Error { + override name = "TursoLockOwnershipLostError"; +} + +export function isTursoLockOwnershipLostError( + error: unknown +): error is TursoLockOwnershipLostError { + return error instanceof TursoLockOwnershipLostError; +} + /** Shared lease check for both write gating and atomic migration acquisition. */ export function readLiveTursoLock(path: string): LockState | null { - if (!existsSync(path)) return null; + let fd: number; + try { + fd = openSync(path, "r"); + } catch { + return null; + } + const identity = fstatSync(fd); let state: LockState | null = null; let reason = "corrupt lock"; try { - state = JSON.parse(readFileSync(path, "utf-8")) as LockState | null; + state = JSON.parse(readFileSync(fd, "utf-8")) as LockState | null; if (state && Number.isInteger(state.pid) && state.pid > 0) { - const age = lockAgeMs(path, state); + const age = lockAgeMs(identity, state); if (isProcessAlive(state.pid)) { if (age === null || age <= TURSO_LOCK_STALE_MS) return state; reason = "progress heartbeat expired"; @@ -66,16 +116,11 @@ export function readLiveTursoLock(path: string): LockState | null { } } catch { // Corrupt locks are stale and removed below. + } finally { + closeSync(fd); } - const ageMs = lockAgeMs(path, state); - let removed = false; - try { - unlinkSync(path); - removed = true; - } catch { - // Already removed (race) or held open (Windows) — either way the lock is - // gone or unowned; do not let cleanup failure block writes. - } + const ageMs = lockAgeMs(identity, state); + const removed = unlinkTursoLockIfUnchanged(path, identity); const details = { path, pid: state?.pid ?? null, @@ -91,11 +136,19 @@ export function readLiveTursoLock(path: string): LockState | null { // turn an unowned lock into a permanent write outage. console.warn("WARNING: Invalidating stale Turso migration lock", details); } + if (!removed) { + try { + // A replacement must be evaluated on its own lease, not the stale read. + if (!sameFile(identity, statSync(path))) return readLiveTursoLock(path); + } catch { + // Another owner already removed it. + } + } return null; } /** Refresh progress only while this process still owns the lease. */ -export function heartbeatTursoLock(path: string): void { +export function heartbeatTursoLock(path: string, expectedId: string): void { let fd: number | undefined; try { // Open without creating, and update the same file we checked. If another @@ -103,7 +156,7 @@ export function heartbeatTursoLock(path: string): void { fd = openSync(path, "r+"); const previous = readFileSync(fd, "utf-8"); const state = JSON.parse(previous) as LockState | null; - if (state?.pid !== process.pid) throw new Error("lock owner changed"); + if (state?.id !== expectedId) throw new Error("lock owner changed"); const updated = JSON.stringify({ ...state, timestamp: new Date().toISOString() }); // Keep any shortened tail as JSON whitespace rather than leaving a corrupt // window between a rewrite and truncate (older locks may be pretty-printed). @@ -116,9 +169,12 @@ export function heartbeatTursoLock(path: string): void { throw new Error("lock file replaced"); } } catch (error) { - throw new Error(`Turso migration lock ownership lost at ${path}; aborting migration`, { - cause: error, - }); + throw new TursoLockOwnershipLostError( + `Turso migration lock ownership lost at ${path}; aborting migration`, + { + cause: error, + } + ); } finally { if (fd !== undefined) closeSync(fd); } @@ -142,20 +198,15 @@ export function acquireTursoOperationLock( ): (() => void) & { heartbeat: () => void } { assertNoTursoMigrationInProgress(); const path = join(CONFIG.storagePath, OPERATION_LOCK); + const id = randomUUID(); const state: LockState = { + id, pid: process.pid, timestamp: new Date().toISOString(), operation, }; writeFileSync(path, JSON.stringify(state), { flag: "wx" }); - const release = () => { - try { - const current = JSON.parse(readFileSync(path, "utf-8")) as LockState | null; - if (current?.pid === process.pid) unlinkSync(path); - } catch { - // Missing, replaced, corrupt, or held open: cleanup must not block writes. - } - }; - return Object.assign(release, { heartbeat: () => heartbeatTursoLock(path) }); + const release = () => releaseTursoLock(path, id); + return Object.assign(release, { heartbeat: () => heartbeatTursoLock(path, id) }); } diff --git a/tests/turso-legacy-migrator.test.ts b/tests/turso-legacy-migrator.test.ts index 2d1d251a..84ba38b3 100644 --- a/tests/turso-legacy-migrator.test.ts +++ b/tests/turso-legacy-migrator.test.ts @@ -191,7 +191,7 @@ describe("turso legacy migrator", () => { expect(JSON.parse(String(row?.vector))).toEqual([1, 0, 0, 0]); }); - for (const loseLock of [false, true]) { + for (const loseLock of [false, true, "same-process"] as const) { migrationTest(`heartbeats insert batches (ownership lost: ${loseLock})`, async () => { baseDir = mkdtempSync(join(tmpdir(), "turso-migrate-heartbeat-")); const projectsDir = join(baseDir, "projects"); @@ -211,7 +211,8 @@ describe("turso legacy migrator", () => { CONFIG.embeddingDimensions = 4; CONFIG.databaseEncryptionEnabled = false; CONFIG.databaseEncryptionKey = undefined; - const { TURSO_LOCK_STALE_MS } = await import("../src/services/turso/operation-lock.js"); + const { TURSO_LOCK_STALE_MS, readLiveTursoLock } = + await import("../src/services/turso/operation-lock.js"); const { tursoVectorSearch } = await import("../src/services/turso/vector-search.js"); const insert = tursoVectorSearch.insertVectorInTransaction; const lockPath = join(baseDir, ".turso-migrate.lock"); @@ -219,13 +220,28 @@ describe("turso legacy migrator", () => { let inserted = 0; tursoVectorSearch.insertVectorInTransaction = async function (tx, record) { if (inserted === 0) { + const lock = JSON.parse(readFileSync(lockPath, "utf-8")); writeFileSync( lockPath, JSON.stringify({ - pid: loseLock ? 2147483647 : process.pid, + ...lock, + id: loseLock === true ? "foreign-acquisition" : lock.id, + pid: loseLock === true ? 2147483647 : process.pid, timestamp: oldTimestamp, }) ); + if (loseLock === "same-process") { + expect(readLiveTursoLock(lockPath)).toBeNull(); + writeFileSync( + lockPath, + JSON.stringify({ + id: "successor", + pid: process.pid, + timestamp: new Date().toISOString(), + }), + { flag: "wx" } + ); + } } if (inserted === 50) { const lock = JSON.parse(readFileSync(lockPath, "utf-8")); @@ -242,7 +258,9 @@ describe("turso legacy migrator", () => { /ownership lost.*aborting migration/ ); expect(existsSync(join(baseDir, ".turso-migrated"))).toBe(false); - expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); + const replacement = JSON.parse(readFileSync(lockPath, "utf-8")); + expect(replacement.pid).toBe(loseLock === true ? 2147483647 : process.pid); + if (loseLock === "same-process") expect(replacement.id).toBe("successor"); } else { await runLegacyTursoMigration(); expect(inserted).toBe(51); diff --git a/tests/turso-operation-lock.test.ts b/tests/turso-operation-lock.test.ts index acdb8456..100862a2 100644 --- a/tests/turso-operation-lock.test.ts +++ b/tests/turso-operation-lock.test.ts @@ -1,5 +1,14 @@ -import { afterEach, describe, expect, it } from "bun:test"; -import { existsSync, mkdtempSync, readFileSync, utimesSync, writeFileSync } from "node:fs"; +import { afterEach, describe, expect, it, spyOn } from "bun:test"; +import * as fs from "node:fs"; +import { + existsSync, + mkdtempSync, + readFileSync, + renameSync, + statSync, + utimesSync, + writeFileSync, +} from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { cleanupTursoTestDirectory } from "./turso-test-utils.js"; @@ -27,6 +36,46 @@ describe("turso operation lock staleness", () => { const oldTimestamp = () => new Date(Date.now() - staleMs - 60_000).toISOString(); + it("does not unlink a replacement file using the inspected stale identity", async () => { + const { unlinkTursoLockIfUnchanged } = await withStorage(); + const path = join(baseDir, ".turso-operation.lock"); + writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp: oldTimestamp() })); + const identity = statSync(path); + renameSync(path, `${path}.old`); + writeFileSync(path, "replacement"); + expect(unlinkTursoLockIfUnchanged(path, identity)).toBe(false); + expect(readFileSync(path, "utf-8")).toBe("replacement"); + }); + + it("re-evaluates and blocks a fresh replacement between stale read and cleanup", async () => { + const { assertNoTursoMigrationInProgress } = await withStorage(); + const path = join(baseDir, ".turso-operation.lock"); + writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp: oldTimestamp() })); + const original = fs.readFileSync; + let replaced = false; + const read = spyOn(fs, "readFileSync").mockImplementation((( + ...args: Parameters + ) => { + const contents = original(...args); + if (typeof args[0] === "number" && !replaced) { + replaced = true; + renameSync(path, `${path}.old`); + writeFileSync( + path, + JSON.stringify({ pid: process.pid, timestamp: new Date().toISOString(), id: "foreign" }) + ); + } + return contents; + }) as typeof original); + try { + expect(() => assertNoTursoMigrationInProgress()).toThrow(/writes are temporarily blocked/); + expect(replaced).toBe(true); + expect(JSON.parse(original(path, "utf-8")).id).toBe("foreign"); + } finally { + read.mockRestore(); + } + }); + for (const file of [".turso-operation.lock", ".turso-migrate.lock"]) { it(`cleans an aged-out live PID in ${file}`, async () => { const { assertNoTursoMigrationInProgress } = await withStorage(); @@ -96,25 +145,43 @@ describe("turso operation lock staleness", () => { expect(existsSync(path)).toBe(false); }); + it("rejects an expired acquisition after same-process takeover", async () => { + const { acquireTursoOperationLock } = await withStorage(); + const first = acquireTursoOperationLock("first"); + const path = join(baseDir, ".turso-operation.lock"); + const state = JSON.parse(readFileSync(path, "utf-8")); + writeFileSync(path, JSON.stringify({ ...state, timestamp: oldTimestamp() })); + const second = acquireTursoOperationLock("second"); + expect(() => first.heartbeat()).toThrow(/ownership lost/); + first(); + expect(JSON.parse(readFileSync(path, "utf-8")).operation).toBe("second"); + second.heartbeat(); + second(); + expect(existsSync(path)).toBe(false); + }); + it("refreshes legacy progress without relying on the acquisition timestamp", async () => { const { heartbeatTursoLock, assertNoTursoMigrationInProgress } = await withStorage(); const path = join(baseDir, ".turso-migrate.lock"); - writeFileSync(path, JSON.stringify({ pid: process.pid, timestamp: oldTimestamp() }, null, 2)); - heartbeatTursoLock(path); + writeFileSync( + path, + JSON.stringify({ id: "legacy-test", pid: process.pid, timestamp: oldTimestamp() }, null, 2) + ); + heartbeatTursoLock(path, "legacy-test"); expect(() => assertNoTursoMigrationInProgress()).toThrow(/writes are temporarily blocked/); }); it("aborts a heartbeat when the lock is missing, corrupt, or no longer ours", async () => { const { heartbeatTursoLock } = await withStorage(); const path = join(baseDir, ".turso-migrate.lock"); - expect(() => heartbeatTursoLock(path)).toThrow(/ownership lost.*aborting migration/); + expect(() => heartbeatTursoLock(path, "test")).toThrow(/ownership lost.*aborting migration/); expect(existsSync(path)).toBe(false); for (const contents of [ "{broken", JSON.stringify({ pid: 2147483647, timestamp: oldTimestamp() }), ]) { writeFileSync(path, contents); - expect(() => heartbeatTursoLock(path)).toThrow(/ownership lost.*aborting migration/); + expect(() => heartbeatTursoLock(path, "test")).toThrow(/ownership lost.*aborting migration/); expect(readFileSync(path, "utf-8")).toBe(contents); } }); diff --git a/tests/turso-reembed-migration.test.ts b/tests/turso-reembed-migration.test.ts index 2b4ea7b0..71f53a05 100644 --- a/tests/turso-reembed-migration.test.ts +++ b/tests/turso-reembed-migration.test.ts @@ -150,6 +150,7 @@ describe("turso re-embed migration safety", () => { lockPath, JSON.stringify({ ...lock, + id: loseLock ? "foreign-acquisition" : lock.id, pid: loseLock ? 2147483647 : process.pid, timestamp: oldTimestamp, }) From 37b125f4a7980fd364d44419fbb42b7fffa341c7 Mon Sep 17 00:00:00 2001 From: Phototonic <35551999+Phototonic@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:22:51 +1100 Subject: [PATCH 5/5] fix(turso): preserve successor data after lock ownership loss --- src/services/memory-portability-service.ts | 13 ++- src/services/shard-path-migration-service.ts | 23 +++- tests/memory-portability.test.ts | 110 +++++++++++++++++++ tests/shard-path-migrate.test.ts | 70 ++++++++++++ 4 files changed, 209 insertions(+), 7 deletions(-) diff --git a/src/services/memory-portability-service.ts b/src/services/memory-portability-service.ts index b31ee8ba..61b54616 100644 --- a/src/services/memory-portability-service.ts +++ b/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"; @@ -381,9 +384,11 @@ export class MemoryPortabilityService { containerTag: target.tag, }; } catch (error) { - // Best-effort cleanup if a partial insert somehow escaped the transaction. - // Heartbeat failures roll back the transaction; those IDs are harmless no-ops here. - 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) { diff --git a/src/services/shard-path-migration-service.ts b/src/services/shard-path-migration-service.ts index 9652bfdf..bbccce86 100644 --- a/src/services/shard-path-migration-service.ts +++ b/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"; @@ -387,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(); diff --git a/tests/memory-portability.test.ts b/tests/memory-portability.test.ts index 9f917cce..22828b9c 100644 --- a/tests/memory-portability.test.ts +++ b/tests/memory-portability.test.ts @@ -196,6 +196,116 @@ describe("memory export/import portability", () => { expect(JSON.parse(readFileSync(lockPath, "utf-8")).pid).toBe(2147483647); }); + for (const failure of ["mid-transaction", "post-commit", "ordinary"] as const) { + const ownershipLoss = failure !== "ordinary"; + it(`handles catch-time rows after ${failure} failure`, async () => { + await stubEmbedding(); + await setupProject(); + const { CONFIG } = await import("../src/config.js"); + const { memoryPortabilityService } = + await import("../src/services/memory-portability-service.js"); + const { tursoShardManager } = await import("../src/services/turso/shard-manager.js"); + const { tursoConnectionManager } = + await import("../src/services/turso/connection-manager.js"); + const { tursoVectorSearch } = await import("../src/services/turso/vector-search.js"); + const { getProjectTagInfo } = await import("../src/services/tags.js"); + const { TURSO_LOCK_STALE_MS } = await import("../src/services/turso/operation-lock.js"); + const inputPath = join(baseDir, "memories.json"); + await memoryPortabilityService.exportMemories({ + currentDirectory: projectDir, + outputPath: inputPath, + }); + const targetDir = join(baseDir, "target"); + mkdirSync(targetDir); + const tag = getProjectTagInfo(targetDir); + const hash = tag.tag.split("_").pop()!; + const shard = await tursoShardManager.createShard("project", hash, 0); + const db = await tursoConnectionManager.getConnection(shard.dbPath); + const lockPath = join(CONFIG.storagePath, ".turso-operation.lock"); + const insert = tursoVectorSearch.insertVectorInTransaction; + const scopeLock = tursoShardManager.withScopeWriteLock; + const setCount = tursoShardManager.setVectorCount; + const document = JSON.parse(readFileSync(inputPath, "utf-8")); + document.memories[0].isPinned = false; + writeFileSync(inputPath, JSON.stringify(document)); + let rolledBack = false; + tursoVectorSearch.insertVectorInTransaction = async function (...args) { + await insert.apply(this, args); + if (failure === "mid-transaction") { + writeFileSync( + lockPath, + JSON.stringify({ + id: "takeover", + pid: process.pid, + timestamp: new Date().toISOString(), + }) + ); + } + }; + // For ordinary failure, fail after commit so compensating cleanup has work. + tursoShardManager.setVectorCount = async function (...args) { + if (!ownershipLoss) throw new Error("ordinary metadata failure"); + if (failure === "post-commit") { + writeFileSync( + lockPath, + JSON.stringify({ + id: "takeover", + pid: process.pid, + timestamp: new Date().toISOString(), + }) + ); + } + return setCount.apply(this, args); + }; + tursoShardManager.withScopeWriteLock = async function (...args) { + try { + return await scopeLock.apply(this, args); + } catch (error) { + tursoShardManager.withScopeWriteLock = scopeLock; + tursoVectorSearch.insertVectorInTransaction = insert; + tursoShardManager.setVectorCount = setCount; + if (failure === "mid-transaction") { + expect(await tursoVectorSearch.countAllVectors(db)).toBe(0); + rolledBack = true; + // Original transaction and scope lock have unwound. A successor can + // now acquire the expired replacement and commit the very same ID. + writeFileSync( + lockPath, + JSON.stringify({ + id: "takeover", + pid: process.pid, + timestamp: new Date(Date.now() - TURSO_LOCK_STALE_MS - 60_000).toISOString(), + }) + ); + const successor = await memoryPortabilityService.importMemories({ + currentDirectory: targetDir, + inputPath, + }); + expect(successor.success).toBe(true); + expect(await tursoVectorSearch.getMemoryById(db, "mem_portable")).toBeTruthy(); + } + throw error; + } + }; + try { + const result = await memoryPortabilityService.importMemories({ + currentDirectory: targetDir, + inputPath, + }); + expect(result.success).toBe(false); + expect(result.error).toContain( + ownershipLoss ? "ownership lost" : "ordinary metadata failure" + ); + expect(rolledBack).toBe(failure === "mid-transaction"); + expect(await tursoVectorSearch.countAllVectors(db)).toBe(ownershipLoss ? 1 : 0); + } finally { + tursoShardManager.withScopeWriteLock = scopeLock; + tursoVectorSearch.insertVectorInTransaction = insert; + tursoShardManager.setVectorCount = setCount; + } + }); + } + it("rejects unsupported schema versions", async () => { await stubEmbedding(); await setupProject(); diff --git a/tests/shard-path-migrate.test.ts b/tests/shard-path-migrate.test.ts index ff7aa450..b62d4ae3 100644 --- a/tests/shard-path-migrate.test.ts +++ b/tests/shard-path-migrate.test.ts @@ -315,6 +315,76 @@ describe("shard path migration", () => { } ); + for (const journalRemains of [false, true]) { + migrationTest( + `preserves swaps and successor memories on ownership loss (journal remains: ${journalRemains})`, + async () => { + const getProjectTagInfo = await createProjects(); + const { hash } = await seedShard(oldProjectDir, [{ id: "original", content: "keep me" }]); + const { CONFIG } = await import("../src/config.js"); + const { tursoShardManager } = await import("../src/services/turso/shard-manager.js"); + const { tursoConnectionManager } = + await import("../src/services/turso/connection-manager.js"); + const { tursoVectorSearch } = await import("../src/services/turso/vector-search.js"); + const originalShard = (await tursoShardManager.getAllShards("project", hash))[0]!; + const reassign = tursoShardManager.reassignShardScope; + const target = getProjectTagInfo(newProjectDir); + const targetHash = target.tag.split("_").pop()!; + let successorPath = ""; + tursoShardManager.reassignShardScope = async function (...args) { + await reassign.apply(this, args); + successorPath = args[2]; + writeFileSync( + join(CONFIG.storagePath, ".turso-operation.lock"), + JSON.stringify({ + id: "successor", + pid: process.pid, + timestamp: new Date().toISOString(), + }) + ); + const db = await tursoConnectionManager.getConnection(successorPath); + await tursoVectorSearch.insertVector(db, { + id: "successor-memory", + content: "written after takeover", + vector: new Float32Array([1, 0]), + containerTag: target.tag, + type: "project", + createdAt: 123, + updatedAt: 123, + }); + if (journalRemains) { + const { heartbeatTursoLock } = await import("../src/services/turso/operation-lock.js"); + heartbeatTursoLock(join(CONFIG.storagePath, ".turso-operation.lock"), "former-owner"); + } + }; + try { + const { shardPathMigrationService } = + await import("../src/services/shard-path-migration-service.js"); + const result = await shardPathMigrationService.migrate({ + currentDirectory: newProjectDir, + fromHash: hash, + allowLinkedSource: true, + }); + expect(result.success).toBe(false); + expect(result.error).toMatch(/interrupted.*may have completed.*recoverable/); + expect(existsSync(join(CONFIG.storagePath, ".path-migrate-swap.json"))).toBe( + journalRemains + ); + expect(existsSync(originalShard.dbPath)).toBe(false); + expect(await tursoShardManager.getAllShards("project", hash)).toHaveLength(0); + expect((await tursoShardManager.getAllShards("project", targetHash))[0]?.dbPath).toBe( + successorPath + ); + const db = await tursoConnectionManager.getConnection(successorPath); + expect(await tursoVectorSearch.getMemoryById(db, "original")).toBeTruthy(); + expect(await tursoVectorSearch.getMemoryById(db, "successor-memory")).toBeTruthy(); + } finally { + tursoShardManager.reassignShardScope = reassign; + } + } + ); + } + migrationTest("refuses to migrate a still-linked source unless explicitly allowed", async () => { await createProjects(); const { hash: oldHash } = await seedShard(oldProjectDir, [