David Snelling
a2f4f6a550
restore() removed the entire live brain directory and then fs.cp'd the snapshot in, so any copy failure left the store destroyed with only a partial copy. The sharpest edge: fs.cp materialized the holes of sparse mmap blob files, so a snapshot that fits on disk could balloon and ENOSPC mid-copy — the recovery tool destroying the brain it was asked to recover. Rewrite restoreFromDirectory as stage → verify → atomic swap: - Copy the snapshot into a _restore_staging area BEFORE touching live data, sparse-aware (copyTreeSparse/copyFileSparse skip all-zero 4 MiB chunks, so a mostly-hole store restores at its true allocated size, not its apparent size). - On any copy failure (ENOSPC included) remove only the half-written staging area and throw — the live store is left exactly as it was. - Only after the copy succeeds, fsync a completion marker naming the staged entries, then swapStagedRestoreIn(): an idempotent per-entry rm-old + rename-staged-in (same-filesystem renames that cannot ENOSPC), removing stale live entries the snapshot lacks. - completeInterruptedRestore(), wired into init() right after the root dir is ensured, finishes a crash mid-swap FORWARD (resume a committed swap) or discards an uncommitted staging area (live still authoritative). _restore_staging is excluded from snapshots. persist() (hard-link snapshot) was already safe and is unchanged; no public API change. Regression (tests/integration/restore-nondestructive.test.ts): a forced copy failure leaves live data fully intact and cleans staging; a normal restore round-trips to the snapshot; an interrupted-but-committed restore completes on reopen; an uncommitted staging area is discarded on open; the sparse copy is byte-identical with allocated blocks far below apparent size.
2514 lines
92 KiB
TypeScript
2514 lines
92 KiB
TypeScript
/**
|
||
* File System Storage Adapter
|
||
* File system storage adapter for Node.js environments
|
||
*/
|
||
|
||
import {
|
||
HNSWNoun,
|
||
HNSWNounWithMetadata,
|
||
StatisticsData,
|
||
NounType
|
||
} from '../../coreTypes.js'
|
||
import {
|
||
BaseStorage,
|
||
StorageBatchConfig,
|
||
SYSTEM_DIR,
|
||
STATISTICS_KEY,
|
||
WriterLockInfo
|
||
} from '../baseStorage.js'
|
||
import { getBrainyVersion } from '../../utils/index.js'
|
||
|
||
// Node.js modules - dynamically imported to avoid issues in browser environments
|
||
let fs: any
|
||
let path: any
|
||
let zlib: any
|
||
let moduleLoadingPromise: Promise<void> | null = null
|
||
|
||
// Try to load Node.js modules
|
||
try {
|
||
// Using dynamic imports to avoid issues in browser environments
|
||
const fsPromise = import('node:fs')
|
||
const pathPromise = import('node:path')
|
||
const zlibPromise = import('node:zlib')
|
||
|
||
moduleLoadingPromise = Promise.all([fsPromise, pathPromise, zlibPromise])
|
||
.then(([fsModule, pathModule, zlibModule]) => {
|
||
fs = fsModule
|
||
path = pathModule.default
|
||
zlib = zlibModule
|
||
})
|
||
.catch((error) => {
|
||
console.error('Failed to load Node.js modules:', error)
|
||
throw error
|
||
})
|
||
} catch (error) {
|
||
console.error(
|
||
'FileSystemStorage: Failed to load Node.js modules. This adapter is not supported in this environment.',
|
||
error
|
||
)
|
||
}
|
||
|
||
/**
|
||
* File system storage adapter for Node.js environments
|
||
* Uses the file system to store data in the specified directory structure
|
||
*
|
||
* Type-aware storage now built into BaseStorage
|
||
* - Removed 10 *_internal method overrides (now inherit from BaseStorage's type-first implementation)
|
||
* - Removed 2 pagination method overrides (getNounsWithPagination, getVerbsWithPagination)
|
||
* - Updated HNSW methods to use BaseStorage's getNoun/saveNoun (type-first paths)
|
||
* - All operations now use type-first paths: entities/nouns/{type}/vectors/{shard}/{id}.json
|
||
*/
|
||
export class FileSystemStorage extends BaseStorage {
|
||
// FileSystem-specific count persistence
|
||
private countsFilePath?: string // Will be set after init
|
||
|
||
// Fixed sharding configuration for optimal balance of simplicity and performance
|
||
// Single-level sharding (depth=1) provides excellent performance for 1-2.5M entities
|
||
// Structure: nouns/ab/uuid.json where 'ab' = first 2 hex chars of UUID
|
||
// - 256 shard directories (00-ff)
|
||
// - Handles 2.5M+ entities with < 10K files per shard
|
||
// - Eliminates dynamic depth changes that cause path mismatch bugs
|
||
private readonly SHARDING_DEPTH = 1 as const
|
||
protected rootDir: string
|
||
private nounsDir!: string
|
||
private verbsDir!: string
|
||
private metadataDir!: string
|
||
private nounMetadataDir!: string
|
||
private verbMetadataDir!: string
|
||
private indexDir!: string // Legacy - for backward compatibility
|
||
private systemDir!: string
|
||
private lockDir!: string
|
||
// Root for raw binary blobs (`<rootDir>/_blobs`). Blobs are stored verbatim
|
||
// (no JSON envelope, no compression) so native code can mmap them directly via
|
||
// getBinaryBlobPath(). Set in init() once the path module is loaded.
|
||
private blobsDir!: string
|
||
private activeLocks: Set<string> = new Set()
|
||
private lockTimers: Map<string, NodeJS.Timeout> = new Map() // Track timers for cleanup
|
||
private allTimers: Set<NodeJS.Timeout> = new Set() // Track all timers for cleanup
|
||
|
||
// Writer-lock state. The writer lock at `locks/_writer.lock` is acquired
|
||
// at Brainy.init() in writer mode and released at close(). A heartbeat
|
||
// timer rewrites the lock every 10s so stale-lock detection can tell a dead
|
||
// writer from a slow one. The constant name matches the file path used.
|
||
private static readonly WRITER_LOCK_FILE = '_writer.lock'
|
||
private static readonly WRITER_HEARTBEAT_MS = 10_000
|
||
private static readonly WRITER_STALE_THRESHOLD_MS = 60_000
|
||
private writerLockHeartbeat?: NodeJS.Timeout
|
||
private writerLockInfo?: WriterLockInfo
|
||
|
||
// Flush-request RPC state. The writer polls `locks/_flush_requests/` for
|
||
// new `.req` files and emits `.ack` files in `locks/_flush_responses/` after
|
||
// flushing. Inspectors call `requestFlushOverFilesystem` to drop a request
|
||
// and wait for the ack. Polling interval is short enough to feel synchronous
|
||
// for operator workflows but doesn't pressure the FS.
|
||
private static readonly FLUSH_REQUEST_DIR = '_flush_requests'
|
||
private static readonly FLUSH_RESPONSE_DIR = '_flush_responses'
|
||
private static readonly FLUSH_WATCH_INTERVAL_MS = 500
|
||
private static readonly FLUSH_POLL_INTERVAL_MS = 100
|
||
private static readonly FLUSH_REQUEST_TTL_MS = 60_000
|
||
private flushWatcherInterval?: NodeJS.Timeout
|
||
private flushWatcherInFlight = false
|
||
private flushWatcherOnRequest?: () => Promise<void>
|
||
|
||
// CRITICAL FIX: Mutex locks for HNSW concurrency control
|
||
// Prevents read-modify-write races during concurrent neighbor updates at scale (1000+ ops)
|
||
// Matches MemoryStorage and OPFSStorage behavior (tested in production)
|
||
private hnswLocks = new Map<string, Promise<void>>()
|
||
|
||
// Compression configuration
|
||
private compressionEnabled: boolean = true // Enable gzip compression by default for 60-80% disk savings
|
||
private compressionLevel: number = 6 // zlib compression level (1-9, default: 6 = balanced)
|
||
|
||
// Transaction durability barrier (see GenerationStorage.beginWriteBarrier).
|
||
// Non-null ONLY between beginWriteBarrier() and flushWriteBarrier(). Because
|
||
// canonical writes are tmp+rename (durable only in the page cache until an
|
||
// fsync), the generation store fsyncs everything a transaction wrote before
|
||
// it advances the generation counter. `writeBarrierPaths` collects the
|
||
// root-relative object paths written; `writeBarrierDeleteDirs` the parent
|
||
// dirs of deleted objects (an unlink is durable only once its directory is
|
||
// fsync'd). Both are null outside a transaction, so the tracking `add`s below
|
||
// are free on the single-op and non-transactional write paths.
|
||
private writeBarrierPaths: Set<string> | null = null
|
||
private writeBarrierDeleteDirs: Set<string> | null = null
|
||
|
||
/**
|
||
* Initialize the storage adapter
|
||
* @param rootDirectory The root directory for storage
|
||
* @param options Optional configuration
|
||
*/
|
||
constructor(
|
||
rootDirectory: string,
|
||
options?: {
|
||
compression?: boolean // Enable gzip compression (default: true)
|
||
compressionLevel?: number // Compression level 1-9 (default: 6)
|
||
}
|
||
) {
|
||
super()
|
||
this.rootDir = rootDirectory
|
||
|
||
// Configure compression
|
||
if (options?.compression !== undefined) {
|
||
this.compressionEnabled = options.compression
|
||
}
|
||
if (options?.compressionLevel !== undefined) {
|
||
this.compressionLevel = Math.min(9, Math.max(1, options.compressionLevel))
|
||
}
|
||
|
||
// Defer path operations until init() when path module is guaranteed to be loaded
|
||
}
|
||
|
||
/**
|
||
* Get FileSystem-optimized batch configuration
|
||
*
|
||
* File system storage is I/O bound but not rate limited:
|
||
* - Large batch sizes (500 items)
|
||
* - No delays needed (0ms)
|
||
* - Moderate concurrency (100 operations) - limited by I/O threads
|
||
* - Parallel processing supported
|
||
*
|
||
* @returns FileSystem-optimized batch configuration
|
||
*/
|
||
public override getBatchConfig(): StorageBatchConfig {
|
||
return {
|
||
maxBatchSize: 500,
|
||
batchDelayMs: 0,
|
||
maxConcurrent: 100,
|
||
supportsParallelWrites: true, // Filesystem handles parallel I/O
|
||
rateLimit: {
|
||
operationsPerSecond: 5000, // Depends on disk speed
|
||
burstCapacity: 2000
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Initialize the storage adapter
|
||
*/
|
||
public override async init(): Promise<void> {
|
||
if (this.isInitialized) {
|
||
return
|
||
}
|
||
|
||
// Wait for module loading to complete
|
||
if (moduleLoadingPromise) {
|
||
try {
|
||
await moduleLoadingPromise
|
||
} catch (error) {
|
||
throw new Error(
|
||
'FileSystemStorage requires a Node.js environment, but `fs` and `path` modules could not be loaded.'
|
||
)
|
||
}
|
||
}
|
||
|
||
// Check if Node.js modules are available
|
||
if (!fs || !path) {
|
||
throw new Error(
|
||
'FileSystemStorage requires a Node.js environment, but `fs` and `path` modules could not be loaded.'
|
||
)
|
||
}
|
||
|
||
try {
|
||
// Initialize directory paths now that path module is loaded
|
||
// Clean directory structure
|
||
this.nounsDir = path.join(this.rootDir, 'entities/nouns/hnsw')
|
||
this.verbsDir = path.join(this.rootDir, 'entities/verbs/hnsw')
|
||
this.metadataDir = path.join(this.rootDir, 'entities/nouns/metadata') // Legacy reference
|
||
this.nounMetadataDir = path.join(this.rootDir, 'entities/nouns/metadata')
|
||
this.verbMetadataDir = path.join(this.rootDir, 'entities/verbs/metadata')
|
||
this.indexDir = path.join(this.rootDir, 'indexes')
|
||
this.systemDir = path.join(this.rootDir, SYSTEM_DIR)
|
||
this.lockDir = path.join(this.rootDir, 'locks')
|
||
this.blobsDir = path.join(this.rootDir, '_blobs')
|
||
|
||
// Create the root directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.rootDir)
|
||
|
||
// Finish any restore interrupted by a crash (resume the staged swap, or
|
||
// discard an uncommitted staging area) BEFORE counts/derived state load,
|
||
// so the rest of startup sees the completed store.
|
||
await this.completeInterruptedRestore()
|
||
|
||
// Create the nouns directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.nounsDir)
|
||
|
||
// Create the verbs directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.verbsDir)
|
||
|
||
// Create the metadata directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.metadataDir)
|
||
|
||
// Create the noun metadata directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.nounMetadataDir)
|
||
|
||
// Create the verb metadata directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.verbMetadataDir)
|
||
|
||
// Create both directories for backward compatibility
|
||
await this.ensureDirectoryExists(this.systemDir)
|
||
// Only create legacy directory if it exists (don't create new legacy dirs)
|
||
if (await this.directoryExists(this.indexDir)) {
|
||
await this.ensureDirectoryExists(this.indexDir)
|
||
}
|
||
|
||
// Create the locks directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.lockDir)
|
||
|
||
// Create the binary blobs directory if it doesn't exist
|
||
await this.ensureDirectoryExists(this.blobsDir)
|
||
|
||
// Initialize count management
|
||
this.countsFilePath = path.join(this.systemDir, 'counts.json')
|
||
await this.initializeCounts()
|
||
|
||
// Boot log: new-vs-established, decided from the canonical layout the
|
||
// database actually reads and writes (`entities/nouns/<shard>/<id>/`)
|
||
// plus the known noun count. The legacy hnsw sharding-depth probe and
|
||
// its depth-migration machinery are gone: the 8.0 write path never
|
||
// populated the directory they inspected, so the probe concluded "new
|
||
// installation" for every store on every boot and the migration branch
|
||
// was unreachable.
|
||
const established =
|
||
this.totalNounCount > 0 || (await this.hasCanonicalEntities())
|
||
console.log(
|
||
established
|
||
? `📁 Using depth ${this.SHARDING_DEPTH} sharding (${this.totalNounCount} entities)`
|
||
: `📁 New installation: using depth ${this.SHARDING_DEPTH} sharding (optimal for 1-2.5M entities)`
|
||
)
|
||
|
||
// Initialize GraphAdjacencyIndex and type statistics
|
||
await super.init()
|
||
} catch (error) {
|
||
console.error('Error initializing FileSystemStorage:', error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Check if a directory exists
|
||
*/
|
||
private async directoryExists(dirPath: string): Promise<boolean> {
|
||
try {
|
||
const stats = await fs.promises.stat(dirPath)
|
||
return stats.isDirectory()
|
||
} catch (error) {
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Ensure a directory exists, creating it if necessary
|
||
*/
|
||
private async ensureDirectoryExists(dirPath: string): Promise<void> {
|
||
try {
|
||
await fs.promises.mkdir(dirPath, { recursive: true })
|
||
} catch (error: any) {
|
||
// Ignore EEXIST error, which means the directory already exists
|
||
if (error.code !== 'EEXIST') {
|
||
throw error
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
/**
|
||
* Primitive operation: Write object to path
|
||
* All metadata operations use this internally via base class routing
|
||
* Supports gzip compression for 60-80% disk savings
|
||
* CRITICAL FIX: Added atomic write pattern to prevent file corruption during concurrent imports
|
||
*/
|
||
protected async writeObjectToPath(pathStr: string, data: any): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const fullPath = path.join(this.rootDir, pathStr)
|
||
await this.ensureDirectoryExists(path.dirname(fullPath))
|
||
|
||
if (this.compressionEnabled) {
|
||
// Write compressed data with .gz extension using atomic pattern
|
||
const compressedPath = `${fullPath}.gz`
|
||
const tempPath = `${compressedPath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}`
|
||
|
||
try {
|
||
// ATOMIC WRITE SEQUENCE:
|
||
// 1. Compress and write to temp file
|
||
const jsonString = JSON.stringify(data, null, 2)
|
||
const compressed = await new Promise<Buffer>((resolve, reject) => {
|
||
zlib.gzip(Buffer.from(jsonString, 'utf-8'), { level: this.compressionLevel }, (err: any, result: Buffer) => {
|
||
if (err) reject(err)
|
||
else resolve(result)
|
||
})
|
||
})
|
||
await fs.promises.writeFile(tempPath, compressed)
|
||
|
||
// 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes)
|
||
await fs.promises.rename(tempPath, compressedPath)
|
||
} catch (error: any) {
|
||
// Clean up temp file on any error
|
||
try {
|
||
await fs.promises.unlink(tempPath)
|
||
} catch (cleanupError) {
|
||
// Ignore cleanup errors
|
||
}
|
||
throw error
|
||
}
|
||
|
||
// Clean up uncompressed file if it exists (migration from uncompressed)
|
||
try {
|
||
await fs.promises.unlink(fullPath)
|
||
} catch (error: any) {
|
||
// Ignore if file doesn't exist
|
||
if (error.code !== 'ENOENT') {
|
||
console.warn(`Failed to remove uncompressed file ${fullPath}:`, error)
|
||
}
|
||
}
|
||
} else {
|
||
// Write uncompressed data using atomic pattern
|
||
const tempPath = `${fullPath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}`
|
||
|
||
try {
|
||
// ATOMIC WRITE SEQUENCE:
|
||
// 1. Write to temp file
|
||
await fs.promises.writeFile(tempPath, JSON.stringify(data, null, 2))
|
||
|
||
// 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes)
|
||
await fs.promises.rename(tempPath, fullPath)
|
||
} catch (error: any) {
|
||
// Clean up temp file on any error
|
||
try {
|
||
await fs.promises.unlink(tempPath)
|
||
} catch (cleanupError) {
|
||
// Ignore cleanup errors
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// Transaction durability barrier: record the write so the generation store
|
||
// can fsync it before advancing the counter. Reached only on a successful
|
||
// rename; the logical (non-`.gz`) path is stored — flushWriteBarrier's
|
||
// syncRawObjects resolves the compressed variant.
|
||
this.writeBarrierPaths?.add(pathStr)
|
||
}
|
||
|
||
/**
|
||
* Primitive operation: Read object from path
|
||
* All metadata operations use this internally via base class routing
|
||
* Enhanced error handling for corrupted metadata files (Bug #3 mitigation)
|
||
* Supports reading both compressed (.gz) and uncompressed files for backward compatibility
|
||
*/
|
||
protected async readObjectFromPath(pathStr: string): Promise<any | null> {
|
||
await this.ensureInitialized()
|
||
|
||
const fullPath = path.join(this.rootDir, pathStr)
|
||
const compressedPath = `${fullPath}.gz`
|
||
|
||
// Try reading compressed file first (if compression is enabled or file exists)
|
||
try {
|
||
const compressedData = await fs.promises.readFile(compressedPath)
|
||
const decompressed = await new Promise<Buffer>((resolve, reject) => {
|
||
zlib.gunzip(compressedData, (err: any, result: Buffer) => {
|
||
if (err) reject(err)
|
||
else resolve(result)
|
||
})
|
||
})
|
||
return JSON.parse(decompressed.toString('utf-8'))
|
||
} catch (error: any) {
|
||
// If compressed file doesn't exist, fall back to uncompressed
|
||
if (error.code !== 'ENOENT') {
|
||
console.warn(`Failed to read compressed file ${compressedPath}:`, error)
|
||
}
|
||
}
|
||
|
||
// Fall back to reading uncompressed file (for backward compatibility)
|
||
try {
|
||
const data = await fs.promises.readFile(fullPath, 'utf-8')
|
||
return JSON.parse(data)
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') {
|
||
return null
|
||
}
|
||
|
||
// Enhanced error handling for corrupted JSON files (race condition from Bug #3)
|
||
if (error instanceof SyntaxError || error.name === 'SyntaxError') {
|
||
console.warn(
|
||
`⚠️ Corrupted metadata file detected: ${pathStr}\n` +
|
||
` This may be caused by concurrent writes during import.\n` +
|
||
` Gracefully skipping this entry. File may be repaired on next write.`
|
||
)
|
||
return null
|
||
}
|
||
|
||
console.error(`Error reading object from ${pathStr}:`, error)
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Primitive operation: Delete object from path
|
||
* All metadata operations use this internally via base class routing
|
||
* Deletes both compressed and uncompressed versions (for cleanup)
|
||
*/
|
||
protected async deleteObjectFromPath(pathStr: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const fullPath = path.join(this.rootDir, pathStr)
|
||
const compressedPath = `${fullPath}.gz`
|
||
|
||
// Try deleting both compressed and uncompressed files (for cleanup during migration)
|
||
let deletedCount = 0
|
||
|
||
// Delete compressed file
|
||
try {
|
||
await fs.promises.unlink(compressedPath)
|
||
deletedCount++
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.warn(`Error deleting compressed file ${compressedPath}:`, error)
|
||
}
|
||
}
|
||
|
||
// Delete uncompressed file
|
||
try {
|
||
await fs.promises.unlink(fullPath)
|
||
deletedCount++
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error deleting uncompressed file ${pathStr}:`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// If neither file existed, it's not an error (already deleted)
|
||
if (deletedCount === 0) {
|
||
// File doesn't exist - this is fine
|
||
}
|
||
|
||
// Transaction durability barrier: an unlink is durable only once its parent
|
||
// directory is fsync'd. Record the dir (root-relative; '.' for a top-level
|
||
// object) so flushWriteBarrier can sync it before the counter advances.
|
||
this.writeBarrierDeleteDirs?.add(path.dirname(pathStr))
|
||
}
|
||
|
||
/**
|
||
* Primitive operation: List objects under path prefix
|
||
* All metadata operations use this internally via base class routing
|
||
* Handles both .json and .json.gz files, normalizes paths
|
||
*/
|
||
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const fullPath = path.join(this.rootDir, prefix)
|
||
const paths: string[] = []
|
||
const seen = new Set<string>() // Track files to avoid duplicates (both .json and .json.gz)
|
||
|
||
try {
|
||
const entries = await fs.promises.readdir(fullPath, { withFileTypes: true })
|
||
|
||
for (const entry of entries) {
|
||
if (entry.isFile()) {
|
||
// Handle multiple compression formats for broad compatibility
|
||
// - .json.gz: Standard entity/metadata files (JSON compressed)
|
||
// - .gz: Raw compressed payloads (e.g. blob-store binary values)
|
||
// - .json: Uncompressed JSON files
|
||
if (entry.name.endsWith('.json.gz')) {
|
||
// Strip .gz extension and add the .json path
|
||
const normalizedName = entry.name.slice(0, -3) // Remove .gz
|
||
const normalizedPath = path.join(prefix, normalizedName)
|
||
if (!seen.has(normalizedPath)) {
|
||
paths.push(normalizedPath)
|
||
seen.add(normalizedPath)
|
||
}
|
||
} else if (entry.name.endsWith('.gz')) {
|
||
// Raw payloads stored as .gz (not .json.gz)
|
||
// Strip .gz extension and return path
|
||
const normalizedName = entry.name.slice(0, -3) // Remove .gz
|
||
const normalizedPath = path.join(prefix, normalizedName)
|
||
if (!seen.has(normalizedPath)) {
|
||
paths.push(normalizedPath)
|
||
seen.add(normalizedPath)
|
||
}
|
||
} else if (entry.name.endsWith('.json')) {
|
||
const filePath = path.join(prefix, entry.name)
|
||
if (!seen.has(filePath)) {
|
||
paths.push(filePath)
|
||
seen.add(filePath)
|
||
}
|
||
}
|
||
} else if (entry.isDirectory()) {
|
||
const subpath = path.join(prefix, entry.name)
|
||
const subdirPaths = await this.listObjectsUnderPath(subpath)
|
||
paths.push(...subdirPaths)
|
||
}
|
||
}
|
||
|
||
return paths.sort()
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') {
|
||
return []
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// ===========================================================================
|
||
// Generational record layer (8.0 MVCC) — durability + snapshot primitives
|
||
// ===========================================================================
|
||
|
||
/**
|
||
* Storage-root-relative paths that are mutated **in place** (appended to)
|
||
* rather than replaced via atomic tmp+rename. `snapshotToDirectory()` must
|
||
* byte-copy these instead of hard-linking them: a hard link shares the
|
||
* inode, so a post-snapshot append to the live file would silently mutate
|
||
* the snapshot. Every other persisted file in this adapter is written via
|
||
* tmp+rename (objects, blobs, counts, locks are excluded entirely), which
|
||
* makes hard links safe — a rewrite swaps in a new inode and the snapshot
|
||
* keeps the old one.
|
||
*/
|
||
private static readonly SNAPSHOT_BYTE_COPY_PATHS = new Set<string>([
|
||
`${SYSTEM_DIR}/tx-log.jsonl`
|
||
])
|
||
|
||
/**
|
||
* Top-level directories whose EVERY file is mutated in place (not tmp+rename)
|
||
* and must therefore be byte-copied into a snapshot, not hard-linked — the
|
||
* directory analogue of {@link SNAPSHOT_BYTE_COPY_PATHS}.
|
||
*
|
||
* `_id_mapper` holds the native provider's shared mmap `BinaryIdMapper`, which
|
||
* `flush()` msyncs and a migration rebuild can truncate+re-inject IN PLACE
|
||
* (confirmed by the native side). A hard link shares the inode, so an in-place
|
||
* msync/truncate on the live file would reach through into the pre-upgrade
|
||
* backup — byte-copy keeps the snapshot a faithful, independent copy. It is
|
||
* bounded (the id map), not the large index files, so the copy cost is small.
|
||
*/
|
||
private static readonly SNAPSHOT_BYTE_COPY_DIRS = new Set<string>(['_id_mapper'])
|
||
|
||
/**
|
||
* Top-level directories excluded from snapshots: process-local lock state
|
||
* (writer lock, flush-request RPC files) must never travel with the data, and
|
||
* the restore staging area ({@link RESTORE_STAGING_DIR}) is transient scratch
|
||
* that must never be captured or restored.
|
||
*/
|
||
private static readonly SNAPSHOT_EXCLUDED_TOP_DIRS = new Set<string>(['locks', '_restore_staging'])
|
||
|
||
/**
|
||
* Transient top-level directory holding a restore-in-progress: the snapshot is
|
||
* fully copied here (sparse-aware) BEFORE any live data is touched, then an
|
||
* atomic per-entry swap moves it into place. Its presence + the completion
|
||
* marker let {@link completeInterruptedRestore} resume a crashed restore.
|
||
*/
|
||
private static readonly RESTORE_STAGING_DIR = '_restore_staging'
|
||
/**
|
||
* Written+fsync'd inside the staging dir ONLY after the whole snapshot has
|
||
* copied successfully. Its presence authorizes the swap (and its resume): a
|
||
* staging dir WITHOUT this marker is an interrupted copy — discardable debris,
|
||
* live data still authoritative.
|
||
*/
|
||
private static readonly RESTORE_MARKER = '.restore-manifest.json'
|
||
/** Chunk size for sparse-aware copying (holes are preserved at this grain). */
|
||
private static readonly SPARSE_CHUNK_BYTES = 4 * 1024 * 1024
|
||
/** A zero buffer the size of one sparse chunk, for all-zero (hole) detection. */
|
||
private static readonly SPARSE_ZERO_CHUNK = Buffer.alloc(4 * 1024 * 1024)
|
||
|
||
/**
|
||
* Remove every object under a storage-root-relative prefix — one recursive
|
||
* directory removal instead of the base class's list+delete loop.
|
||
*
|
||
* @param prefix - Storage-root-relative directory prefix to remove.
|
||
*/
|
||
public override async removeRawPrefix(prefix: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
await fs.promises.rm(path.join(this.rootDir, prefix), { recursive: true, force: true })
|
||
}
|
||
|
||
/**
|
||
* Durability barrier: `fsync` each listed object file (resolving the
|
||
* compressed `.gz` variant when present) and then the set of parent
|
||
* directories, so both the file contents and the rename directory entries
|
||
* are durable before the commit protocol proceeds. Paths whose file no
|
||
* longer exists are skipped (a later step may have replaced them).
|
||
*
|
||
* @param paths - Storage-root-relative object paths previously written.
|
||
*/
|
||
public override async syncRawObjects(paths: string[]): Promise<void> {
|
||
await this.ensureInitialized()
|
||
const parentDirs = new Set<string>()
|
||
|
||
for (const objectPath of paths) {
|
||
const fullPath = path.join(this.rootDir, objectPath)
|
||
for (const candidate of [`${fullPath}.gz`, fullPath]) {
|
||
let handle: any
|
||
try {
|
||
handle = await fs.promises.open(candidate, 'r')
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') continue
|
||
throw error
|
||
}
|
||
try {
|
||
await handle.sync()
|
||
} finally {
|
||
await handle.close()
|
||
}
|
||
parentDirs.add(path.dirname(fullPath))
|
||
break
|
||
}
|
||
}
|
||
|
||
for (const dir of parentDirs) {
|
||
let handle: any
|
||
try {
|
||
handle = await fs.promises.open(dir, 'r')
|
||
} catch {
|
||
continue // directory vanished or platform disallows opening dirs
|
||
}
|
||
try {
|
||
await handle.sync()
|
||
} catch {
|
||
// Some platforms (and some filesystems) reject directory fsync —
|
||
// file-level fsync above already covers the data itself.
|
||
} finally {
|
||
await handle.close()
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Begin a transaction durability barrier: start recording every canonical
|
||
* object write and delete so {@link flushWriteBarrier} can fsync them before
|
||
* the generation counter advances. Resets unconditionally, discarding any
|
||
* tracking left by a transaction that aborted without flushing.
|
||
*
|
||
* @see GenerationStorage.beginWriteBarrier
|
||
*/
|
||
public beginWriteBarrier(): void {
|
||
this.writeBarrierPaths = new Set()
|
||
this.writeBarrierDeleteDirs = new Set()
|
||
}
|
||
|
||
/**
|
||
* Flush the transaction durability barrier: fsync every canonical write since
|
||
* {@link beginWriteBarrier} (file contents AND the rename directory entries,
|
||
* via {@link syncRawObjects}), then fsync the parent directory of every
|
||
* canonical delete so the unlinks are durable too. Clears the tracking. After
|
||
* this resolves, the transaction's entire canonical footprint is on disk, so
|
||
* the generation counter/manifest can be advanced without risking a
|
||
* counter-ahead-of-state torn store on a hard kill.
|
||
*
|
||
* @see GenerationStorage.flushWriteBarrier
|
||
*/
|
||
public async flushWriteBarrier(): Promise<void> {
|
||
const paths = this.writeBarrierPaths
|
||
const deleteDirs = this.writeBarrierDeleteDirs
|
||
this.writeBarrierPaths = null
|
||
this.writeBarrierDeleteDirs = null
|
||
|
||
if (paths && paths.size > 0) {
|
||
// syncRawObjects fsyncs each file and its parent directory.
|
||
await this.syncRawObjects([...paths])
|
||
}
|
||
|
||
if (deleteDirs && deleteDirs.size > 0) {
|
||
for (const relDir of deleteDirs) {
|
||
const dirPath = path.join(this.rootDir, relDir)
|
||
let handle: any
|
||
try {
|
||
handle = await fs.promises.open(dirPath, 'r')
|
||
} catch {
|
||
continue // directory vanished or platform disallows opening dirs
|
||
}
|
||
try {
|
||
await handle.sync()
|
||
} catch {
|
||
// Some platforms/filesystems reject directory fsync — best effort.
|
||
} finally {
|
||
await handle.close()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Append one line to `_system/tx-log.jsonl`. Plain `appendFile` — the
|
||
* tx-log is the one append-in-place file in the store (and is byte-copied,
|
||
* never hard-linked, by {@link FileSystemStorage.snapshotToDirectory}).
|
||
*
|
||
* @param line - One complete JSON document, without trailing newline.
|
||
*/
|
||
public async appendTxLogLine(line: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
const logPath = path.join(this.systemDir, 'tx-log.jsonl')
|
||
await fs.promises.appendFile(logPath, `${line}\n`, 'utf-8')
|
||
}
|
||
|
||
/**
|
||
* Read all tx-log lines, oldest first (empty array when no log exists).
|
||
* Torn trailing lines from a crashed append are returned as-is — callers
|
||
* tolerate unparseable lines.
|
||
*/
|
||
public async readTxLogLines(): Promise<string[]> {
|
||
await this.ensureInitialized()
|
||
const logPath = path.join(this.systemDir, 'tx-log.jsonl')
|
||
try {
|
||
const content: string = await fs.promises.readFile(logPath, 'utf-8')
|
||
return content.split('\n').filter((l: string) => l.length > 0)
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') return []
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Snapshot the entire store into `targetPath` as a hard-link farm
|
||
* (Cassandra-style: instant, space-shared). Safe because every data file
|
||
* is immutable-by-rename — rewrites swap in new inodes, leaving the
|
||
* snapshot's links pointing at the old bytes. The two exceptions are
|
||
* handled explicitly: append-in-place files
|
||
* ({@link FileSystemStorage.SNAPSHOT_BYTE_COPY_PATHS}) are byte-copied, and
|
||
* process-local lock state
|
||
* ({@link FileSystemStorage.SNAPSHOT_EXCLUDED_TOP_DIRS}) is excluded.
|
||
* Cross-device targets (where `link(2)` fails with `EXDEV`) fall back to
|
||
* byte copies per file.
|
||
*
|
||
* @param targetPath - Absolute directory for the snapshot. Created if
|
||
* missing; must be empty or absent (refuses to overwrite).
|
||
*/
|
||
public async snapshotToDirectory(targetPath: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
try {
|
||
const existing = await fs.promises.readdir(targetPath)
|
||
if (existing.length > 0) {
|
||
throw new Error(
|
||
`snapshotToDirectory: target ${targetPath} already exists and is not empty. ` +
|
||
`Choose a fresh directory per snapshot.`
|
||
)
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') throw error
|
||
}
|
||
await fs.promises.mkdir(targetPath, { recursive: true })
|
||
|
||
const files: string[] = []
|
||
await this.collectSnapshotFiles(this.rootDir, '', files)
|
||
|
||
for (const relPath of files) {
|
||
const sourceFile = path.join(this.rootDir, relPath)
|
||
const targetFile = path.join(targetPath, relPath)
|
||
await fs.promises.mkdir(path.dirname(targetFile), { recursive: true })
|
||
|
||
// Byte-copy list: compare against the normalized (extension-preserving)
|
||
// relative path with separators unified, so `_system/tx-log.jsonl`
|
||
// matches on every platform.
|
||
const normalized = relPath.split(path.sep).join('/')
|
||
// Byte-copy (never hard-link) files that are mutated in place: the exact
|
||
// append-in-place paths, and every file under a mmap-mutated directory
|
||
// (e.g. the native `_id_mapper/*`). A shared inode would let a live
|
||
// msync/truncate reach through into the snapshot.
|
||
if (
|
||
FileSystemStorage.SNAPSHOT_BYTE_COPY_PATHS.has(normalized) ||
|
||
FileSystemStorage.SNAPSHOT_BYTE_COPY_DIRS.has(normalized.split('/')[0])
|
||
) {
|
||
await fs.promises.copyFile(sourceFile, targetFile)
|
||
continue
|
||
}
|
||
|
||
try {
|
||
await fs.promises.link(sourceFile, targetFile)
|
||
} catch (error: any) {
|
||
// EXDEV: cross-device; EPERM/ENOTSUP: filesystem forbids links.
|
||
// ENOENT: the live file was atomically replaced mid-walk — retry as
|
||
// a copy of whatever is current (single-writer discipline means this
|
||
// only happens for derived files being flushed concurrently).
|
||
if (['EXDEV', 'EPERM', 'ENOTSUP', 'ENOENT'].includes(error.code)) {
|
||
try {
|
||
await fs.promises.copyFile(sourceFile, targetFile)
|
||
} catch (copyError: any) {
|
||
if (copyError.code !== 'ENOENT') throw copyError
|
||
}
|
||
} else {
|
||
throw error
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @description Pre-upgrade backup: a hard-link snapshot of the whole store into
|
||
* a SIBLING directory (`<rootDir>.migration-backup`, outside `rootDir` so the
|
||
* snapshot never recurses into itself), taken before a 7.x → 8.0 upgrade
|
||
* rebuilds the derived indexes. Zero-copy (shared inodes; the store is
|
||
* immutable-by-rename) and instant even at scale. Returns the backup path, or
|
||
* `null` when the store holds no canonical nouns (nothing to protect). If a
|
||
* backup already exists from a prior FAILED upgrade, it is REUSED as-is (that
|
||
* is the true pre-upgrade state; a retry must not overwrite it).
|
||
*/
|
||
public async createMigrationBackup(): Promise<string | null> {
|
||
await this.ensureInitialized()
|
||
|
||
// Nothing to protect if the store has no canonical nouns (e.g. a brand-new
|
||
// brain whose marker is simply absent — `epochStale` is true but there is no
|
||
// 7.x data to migrate).
|
||
const probe = await this.getNouns({ pagination: { limit: 1 } })
|
||
if ((probe.totalCount || 0) === 0 && probe.items.length === 0) {
|
||
return null
|
||
}
|
||
|
||
const backupPath = `${this.rootDir}.migration-backup`
|
||
|
||
// Reuse an existing backup (a prior failed upgrade's pre-state) rather than
|
||
// overwrite it — snapshotToDirectory also refuses a non-empty target.
|
||
try {
|
||
const existing = await fs.promises.readdir(backupPath)
|
||
if (existing.length > 0) return backupPath
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') throw error
|
||
}
|
||
|
||
await this.snapshotToDirectory(backupPath)
|
||
return backupPath
|
||
}
|
||
|
||
/**
|
||
* @description Remove a {@link createMigrationBackup} snapshot. Best-effort: a
|
||
* missing path is a no-op, and any error is swallowed (a leftover backup dir is
|
||
* harmless — the operator can delete it). Removing the hard-links never touches
|
||
* the live store's bytes (shared inodes; only the extra links go away).
|
||
*/
|
||
public async removeMigrationBackup(location: string): Promise<void> {
|
||
try {
|
||
await fs.promises.rm(location, { recursive: true, force: true })
|
||
} catch {
|
||
// best-effort — leaving the backup behind is safe
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Recursively collect snapshot-eligible files under `dirAbs`, excluding the
|
||
* lock directory and in-flight `*.tmp.*` write files.
|
||
*/
|
||
private async collectSnapshotFiles(dirAbs: string, relPrefix: string, out: string[]): Promise<void> {
|
||
let entries: any[]
|
||
try {
|
||
entries = await fs.promises.readdir(dirAbs, { withFileTypes: true })
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') return
|
||
throw error
|
||
}
|
||
|
||
for (const entry of entries) {
|
||
const rel = relPrefix ? path.join(relPrefix, entry.name) : entry.name
|
||
if (entry.isDirectory()) {
|
||
if (relPrefix === '' && FileSystemStorage.SNAPSHOT_EXCLUDED_TOP_DIRS.has(entry.name)) {
|
||
continue
|
||
}
|
||
await this.collectSnapshotFiles(path.join(dirAbs, entry.name), rel, out)
|
||
} else if (entry.isFile()) {
|
||
if (entry.name.includes('.tmp.')) continue // in-flight atomic write
|
||
out.push(rel)
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Replace the store's contents from a snapshot directory: every current
|
||
* top-level entry except `locks/` (the live writer lock must survive) is
|
||
* removed, the snapshot is byte-copied in (`fs.cp` — never hard-linked, so
|
||
* the snapshot stays independent of the restored store), and all
|
||
* adapter-internal derived state is reloaded.
|
||
*
|
||
* @param sourcePath - Absolute path of a directory produced by
|
||
* {@link FileSystemStorage.snapshotToDirectory}.
|
||
*
|
||
* Non-destructive: the snapshot is copied into a staging area (sparse-aware,
|
||
* so a store of mostly-hole mmap blobs cannot balloon and ENOSPC) BEFORE any
|
||
* live data is touched. Only once the full copy has succeeded and a completion
|
||
* marker is fsync'd does an atomic per-entry swap move it into place. A copy
|
||
* failure (including ENOSPC) leaves the live store exactly as it was; a crash
|
||
* mid-swap is resumed forward on the next {@link init} by
|
||
* {@link completeInterruptedRestore}. The previous implementation removed the
|
||
* live store first and then `fs.cp`'d — a copy failure destroyed the brain it
|
||
* was meant to recover.
|
||
*/
|
||
public async restoreFromDirectory(sourcePath: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const sourceStat = await fs.promises.stat(sourcePath).catch(() => null)
|
||
if (!sourceStat || !sourceStat.isDirectory()) {
|
||
throw new Error(`restoreFromDirectory: ${sourcePath} is not a directory`)
|
||
}
|
||
|
||
const staging = path.join(this.rootDir, FileSystemStorage.RESTORE_STAGING_DIR)
|
||
|
||
// Discard any staging left by a prior aborted restore, then stage fresh.
|
||
await fs.promises.rm(staging, { recursive: true, force: true })
|
||
await fs.promises.mkdir(staging, { recursive: true })
|
||
|
||
// Copy the snapshot into staging (sparse-aware). ANY failure here — most
|
||
// importantly ENOSPC — leaves the live store untouched: we remove only the
|
||
// half-written staging area and re-throw.
|
||
const staged: string[] = []
|
||
try {
|
||
const sourceEntries = await fs.promises.readdir(sourcePath)
|
||
for (const entry of sourceEntries) {
|
||
if (FileSystemStorage.SNAPSHOT_EXCLUDED_TOP_DIRS.has(entry)) continue
|
||
await this.copyTreeSparse(
|
||
path.join(sourcePath, entry),
|
||
path.join(staging, entry)
|
||
)
|
||
staged.push(entry)
|
||
}
|
||
// Commit point: fsync a marker naming the fully-staged entries. Only after
|
||
// this does the swap (and its resume) become authorized.
|
||
const markerPath = path.join(staging, FileSystemStorage.RESTORE_MARKER)
|
||
await fs.promises.writeFile(markerPath, JSON.stringify({ entries: staged }))
|
||
const mfh = await fs.promises.open(markerPath, 'r')
|
||
try {
|
||
await mfh.sync()
|
||
} finally {
|
||
await mfh.close()
|
||
}
|
||
const dfh = await fs.promises.open(staging, 'r').catch(() => null)
|
||
if (dfh) {
|
||
try {
|
||
await dfh.sync()
|
||
} catch {
|
||
// platform may reject directory fsync — best effort
|
||
} finally {
|
||
await dfh.close()
|
||
}
|
||
}
|
||
} catch (error: any) {
|
||
await fs.promises.rm(staging, { recursive: true, force: true }).catch(() => {})
|
||
throw new Error(
|
||
`restoreFromDirectory: staging copy failed, live store left untouched: ${error.message}`
|
||
)
|
||
}
|
||
|
||
// Atomic per-entry swap (metadata-only renames within rootDir — cannot ENOSPC).
|
||
await this.swapStagedRestoreIn()
|
||
await this.reloadDerivedState()
|
||
}
|
||
|
||
/**
|
||
* Move a fully-staged, marker-committed restore into place, then clear the
|
||
* staging area. Idempotent and resumable: driven by the marker's entry list
|
||
* and by which staged entries remain, so a crash at any point is completed by
|
||
* simply calling it again (from {@link completeInterruptedRestore} on the next
|
||
* open). Every step is a same-filesystem rename or a remove — no operation can
|
||
* fail for disk space, so once the marker exists the store is guaranteed to
|
||
* reach the restored state.
|
||
*/
|
||
private async swapStagedRestoreIn(): Promise<void> {
|
||
const staging = path.join(this.rootDir, FileSystemStorage.RESTORE_STAGING_DIR)
|
||
const markerPath = path.join(staging, FileSystemStorage.RESTORE_MARKER)
|
||
|
||
let staged: string[]
|
||
try {
|
||
const marker = JSON.parse(await fs.promises.readFile(markerPath, 'utf-8'))
|
||
staged = Array.isArray(marker?.entries) ? marker.entries : []
|
||
} catch {
|
||
return // no committed marker — nothing to swap
|
||
}
|
||
const stagedSet = new Set(staged)
|
||
|
||
// 1. Remove stale live entries the snapshot does not contain (excluding
|
||
// process-local dirs and the staging area itself). An already-placed
|
||
// staged entry is in stagedSet, so it is kept.
|
||
for (const entry of await fs.promises.readdir(this.rootDir)) {
|
||
if (FileSystemStorage.SNAPSHOT_EXCLUDED_TOP_DIRS.has(entry)) continue
|
||
if (entry === FileSystemStorage.RESTORE_STAGING_DIR) continue
|
||
if (stagedSet.has(entry)) continue
|
||
await fs.promises.rm(path.join(this.rootDir, entry), { recursive: true, force: true })
|
||
}
|
||
|
||
// 2. Place each staged entry (idempotent: a prior attempt that already moved
|
||
// it leaves staging/entry absent, so we skip). `rm` the old first — a
|
||
// rename onto an existing non-empty directory is not allowed; the staged
|
||
// copy is the durable source until it is placed, so a crash between the
|
||
// rm and the rename is recovered forward on the next call.
|
||
for (const entry of staged) {
|
||
const from = path.join(staging, entry)
|
||
const to = path.join(this.rootDir, entry)
|
||
const exists = await fs.promises.lstat(from).then(() => true, () => false)
|
||
if (!exists) continue
|
||
await fs.promises.rm(to, { recursive: true, force: true })
|
||
await fs.promises.rename(from, to)
|
||
}
|
||
|
||
// 3. Clear the staging area (marker last-standing entry).
|
||
await fs.promises.rm(staging, { recursive: true, force: true })
|
||
}
|
||
|
||
/**
|
||
* On open, finish any restore interrupted by a crash. A staging dir WITH the
|
||
* completion marker means the copy had succeeded — resume the swap forward
|
||
* (loudly). A staging dir WITHOUT the marker is an interrupted copy — pure
|
||
* debris; the live store is authoritative, so discard it. Called from
|
||
* {@link init} before counts/derived state load, so recovery is invisible to
|
||
* the rest of startup. Returns `true` if a swap was resumed.
|
||
*/
|
||
private async completeInterruptedRestore(): Promise<boolean> {
|
||
const staging = path.join(this.rootDir, FileSystemStorage.RESTORE_STAGING_DIR)
|
||
const stagingStat = await fs.promises.stat(staging).catch(() => null)
|
||
if (!stagingStat || !stagingStat.isDirectory()) return false
|
||
|
||
const markerPath = path.join(staging, FileSystemStorage.RESTORE_MARKER)
|
||
const hasMarker = await fs.promises
|
||
.stat(markerPath)
|
||
.then(() => true, () => false)
|
||
|
||
if (!hasMarker) {
|
||
// Interrupted before the copy committed — live data untouched, discard.
|
||
await fs.promises.rm(staging, { recursive: true, force: true }).catch(() => {})
|
||
return false
|
||
}
|
||
|
||
console.log('♻️ Resuming an interrupted restore (completing the staged swap)')
|
||
await this.swapStagedRestoreIn()
|
||
return true
|
||
}
|
||
|
||
/**
|
||
* Recursively copy `src` to `dest`, preserving holes (sparse regions). Files
|
||
* are copied chunk-by-chunk skipping all-zero chunks, so a store of
|
||
* mostly-hole mmap blobs restores at its true allocated size instead of
|
||
* materializing every hole (the failure that made `fs.cp` ENOSPC a restore
|
||
* that would otherwise fit). Directories recurse; symlinks are recreated.
|
||
*/
|
||
private async copyTreeSparse(src: string, dest: string): Promise<void> {
|
||
const stat = await fs.promises.lstat(src)
|
||
if (stat.isDirectory()) {
|
||
await fs.promises.mkdir(dest, { recursive: true })
|
||
for (const child of await fs.promises.readdir(src)) {
|
||
await this.copyTreeSparse(path.join(src, child), path.join(dest, child))
|
||
}
|
||
} else if (stat.isSymbolicLink()) {
|
||
await fs.promises.symlink(await fs.promises.readlink(src), dest)
|
||
} else if (stat.isFile()) {
|
||
await this.copyFileSparse(src, dest, stat.size, stat.mode)
|
||
}
|
||
// Other node types (sockets, devices) do not occur in a brain store.
|
||
}
|
||
|
||
/** Sparse-aware single-file copy — see {@link copyTreeSparse}. */
|
||
private async copyFileSparse(
|
||
src: string,
|
||
dest: string,
|
||
size: number,
|
||
mode: number
|
||
): Promise<void> {
|
||
const CHUNK = FileSystemStorage.SPARSE_CHUNK_BYTES
|
||
const srcFh = await fs.promises.open(src, 'r')
|
||
try {
|
||
const destFh = await fs.promises.open(dest, 'w', mode)
|
||
try {
|
||
// Pre-size the destination so unwritten regions are holes.
|
||
await destFh.truncate(size)
|
||
const buf = Buffer.allocUnsafe(CHUNK)
|
||
let pos = 0
|
||
while (pos < size) {
|
||
const { bytesRead } = await srcFh.read(buf, 0, CHUNK, pos)
|
||
if (bytesRead === 0) break
|
||
const chunk = buf.subarray(0, bytesRead)
|
||
// Skip all-zero chunks: leaving them unwritten preserves the hole.
|
||
const isHole = chunk.equals(
|
||
FileSystemStorage.SPARSE_ZERO_CHUNK.subarray(0, bytesRead)
|
||
)
|
||
if (!isHole) {
|
||
await destFh.write(chunk, 0, bytesRead, pos)
|
||
}
|
||
pos += bytesRead
|
||
}
|
||
} finally {
|
||
await destFh.close()
|
||
}
|
||
} finally {
|
||
await srcFh.close()
|
||
}
|
||
}
|
||
|
||
// ===========================================================================
|
||
// Raw binary-blob primitive (mmap-friendly)
|
||
// ===========================================================================
|
||
|
||
/**
|
||
* Resolve a blob key to its on-disk path under `<rootDir>/_blobs`.
|
||
*
|
||
* The key's "/"-separated segments become nested directories and the file is
|
||
* suffixed with `.bin`, e.g. `"graph-lsm/source/sstable-123"` →
|
||
* `<rootDir>/_blobs/graph-lsm/source/sstable-123.bin`. This convention is
|
||
* shared with cor's `MmapFileSystemStorage` so native code and the TS layer
|
||
* agree on exactly where each blob lives.
|
||
*
|
||
* @param key - The blob key.
|
||
* @returns The absolute on-disk path for the blob.
|
||
* @private
|
||
*/
|
||
private blobPath(key: string): string {
|
||
// `path` and `blobsDir` are populated in init(). If a caller resolves a blob
|
||
// path before init() (e.g. native code probing for a mmap target), fall back
|
||
// to a POSIX-style join off rootDir so this synchronous method never throws.
|
||
if (path && this.blobsDir) {
|
||
return path.join(this.blobsDir, ...key.split('/')) + '.bin'
|
||
}
|
||
return `${this.rootDir}/_blobs/${key}.bin`
|
||
}
|
||
|
||
/**
|
||
* Persist a raw binary blob verbatim under `key`, using an atomic
|
||
* temp-file + rename so concurrent readers never observe a torn write.
|
||
* Parent directories are created on demand.
|
||
*
|
||
* @param key - The blob key (see {@link getBinaryBlobPath} for the convention).
|
||
* @param data - The exact bytes to store.
|
||
*/
|
||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||
await this.ensureInitialized()
|
||
const filePath = this.blobPath(key)
|
||
await this.ensureDirectoryExists(path.dirname(filePath))
|
||
// Unique per-writer temp suffix — matches the pattern used at every other
|
||
// atomic-write site in this file (lines 336, 551, 744, 781, 1529, 2908).
|
||
// Without a unique suffix, two concurrent saveBinaryBlob() calls for the
|
||
// same key collide on `${filePath}.tmp`: both writeFile, the first rename
|
||
// succeeds, the second fires against a missing temp and throws ENOENT.
|
||
// Reproduced in production: column-store compaction running alongside an
|
||
// explicit flush() repeatedly raced on `_column_index/<field>/DELETED.bin`.
|
||
const tmpPath = `${filePath}.tmp.${process.pid}.${Date.now()}.${Math.random().toString(36).slice(2)}`
|
||
await fs.promises.writeFile(tmpPath, data)
|
||
try {
|
||
await fs.promises.rename(tmpPath, filePath)
|
||
} catch (err) {
|
||
const code = (err as NodeJS.ErrnoException).code
|
||
// The writes are idempotent for a given key — every caller persists the
|
||
// same logical bytes for that key — so ENOENT on rename means the temp
|
||
// is already gone (rare with the unique suffix, but defensive against
|
||
// crash-resume cleanup paths and external file-system sweepers).
|
||
if (code === 'ENOENT') return
|
||
// Any other failure: clean up our own temp to avoid orphans before rethrow.
|
||
await fs.promises.unlink(tmpPath).catch(() => {})
|
||
throw err
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Load the raw bytes stored under `key`, or `null` if the blob does not exist.
|
||
*
|
||
* @param key - The blob key.
|
||
* @returns The blob bytes, or `null` if absent.
|
||
*/
|
||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||
await this.ensureInitialized()
|
||
try {
|
||
return await fs.promises.readFile(this.blobPath(key))
|
||
} catch {
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Delete the blob stored under `key`. Missing blobs are ignored.
|
||
*
|
||
* @param key - The blob key.
|
||
*/
|
||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
try {
|
||
await fs.promises.unlink(this.blobPath(key))
|
||
} catch {
|
||
/* ignore missing files */
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Return the real on-disk path for `key` so native code can mmap the file
|
||
* directly. The path is returned whether or not the file currently exists —
|
||
* callers are expected to write before mapping.
|
||
*
|
||
* @param key - The blob key.
|
||
* @returns The absolute on-disk path (never `null` for filesystem storage).
|
||
*/
|
||
public getBinaryBlobPath(key: string): string | null {
|
||
return this.blobPath(key)
|
||
}
|
||
|
||
/**
|
||
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
|
||
* FileSystem implementation uses controlled concurrency to prevent too many file reads
|
||
*/
|
||
public async getMetadataBatch(ids: string[]): Promise<Map<string, any>> {
|
||
await this.ensureInitialized()
|
||
|
||
const results = new Map<string, any>()
|
||
const batchSize = 10 // Process 10 files at a time
|
||
|
||
// Process in batches to avoid overwhelming the filesystem
|
||
for (let i = 0; i < ids.length; i += batchSize) {
|
||
const batch = ids.slice(i, i + batchSize)
|
||
|
||
const batchPromises = batch.map(async (id) => {
|
||
try {
|
||
// CRITICAL: Use getNounMetadata() instead of deprecated getMetadata()
|
||
// This ensures we fetch from the correct noun metadata store (2-file system)
|
||
const metadata = await this.getNounMetadata(id)
|
||
return { id, metadata }
|
||
} catch (error) {
|
||
console.debug(`Failed to read metadata for ${id}:`, error)
|
||
return { id, metadata: null }
|
||
}
|
||
})
|
||
|
||
const batchResults = await Promise.all(batchPromises)
|
||
|
||
for (const { id, metadata } of batchResults) {
|
||
if (metadata !== null) {
|
||
results.set(id, metadata)
|
||
}
|
||
}
|
||
|
||
// Small yield between batches
|
||
await new Promise(resolve => setImmediate(resolve))
|
||
}
|
||
|
||
return results
|
||
}
|
||
|
||
/**
|
||
* Get nouns with pagination support
|
||
* @param options Pagination options
|
||
*/
|
||
// Removed getNounsWithPagination override - now uses BaseStorage's type-first implementation
|
||
|
||
/**
|
||
* Clear all data from storage
|
||
*/
|
||
public async clear(): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check if fs module is available
|
||
if (!fs || !fs.promises) {
|
||
console.warn('FileSystemStorage.clear: fs module not available, skipping clear operation')
|
||
return
|
||
}
|
||
|
||
// Helper function to remove all files in a directory
|
||
const removeDirectoryContents = async (dirPath: string): Promise<void> => {
|
||
try {
|
||
const files = await fs.promises.readdir(dirPath)
|
||
for (const file of files) {
|
||
const filePath = path.join(dirPath, file)
|
||
const stats = await fs.promises.stat(filePath)
|
||
if (stats.isDirectory()) {
|
||
await removeDirectoryContents(filePath)
|
||
await fs.promises.rmdir(filePath)
|
||
} else {
|
||
await fs.promises.unlink(filePath)
|
||
}
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error removing directory contents ${dirPath}:`, error)
|
||
throw error
|
||
}
|
||
}
|
||
}
|
||
|
||
// Clear the canonical entity/verb data area
|
||
const entitiesDir = path.join(this.rootDir, 'entities')
|
||
if (await this.directoryExists(entitiesDir)) {
|
||
await removeDirectoryContents(entitiesDir)
|
||
}
|
||
|
||
// Remove all files in both system directories
|
||
await removeDirectoryContents(this.systemDir)
|
||
if (await this.directoryExists(this.indexDir)) {
|
||
await removeDirectoryContents(this.indexDir)
|
||
}
|
||
|
||
// Remove the content-addressed blob store (VFS file content)
|
||
const casDir = path.join(this.rootDir, '_cas')
|
||
if (await this.directoryExists(casDir)) {
|
||
// Delete the entire _cas/ directory (not just contents)
|
||
await fs.promises.rm(casDir, { recursive: true, force: true })
|
||
}
|
||
|
||
// Reset ALL shared derived state via the same path restore uses: the
|
||
// write-through cache, id→type/subtype caches, per-type/subtype count
|
||
// rollups, statistics cache, graph-index singleton, and the BlobStorage
|
||
// instance (its LRU cache holds deleted blobs), then recount from the
|
||
// now-empty data area. Without the rollup reset, stats() would keep
|
||
// reporting per-type counts for deleted entities.
|
||
await this.reloadDerivedState()
|
||
}
|
||
|
||
/**
|
||
* Enhanced clear operation with safety mechanisms and performance optimizations
|
||
* Provides progress tracking, backup options, and instance name confirmation
|
||
*/
|
||
public async clearEnhanced(options: import('../enhancedClearOperations.js').ClearOptions = {}): Promise<import('../enhancedClearOperations.js').ClearResult> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check if fs module is available
|
||
if (!fs || !fs.promises) {
|
||
throw new Error('FileSystemStorage.clearEnhanced: fs module not available')
|
||
}
|
||
|
||
const { EnhancedFileSystemClear } = await import('../enhancedClearOperations.js')
|
||
const enhancedClear = new EnhancedFileSystemClear(this.rootDir, fs, path)
|
||
|
||
const result = await enhancedClear.clear(options)
|
||
|
||
if (result.success) {
|
||
// Clear the statistics cache
|
||
this.statisticsCache = null
|
||
this.statisticsModified = false
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* Get information about storage usage and capacity
|
||
*/
|
||
public async getStorageStatus(): Promise<{
|
||
type: string
|
||
used: number
|
||
quota: number | null
|
||
details?: Record<string, any>
|
||
}> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check if fs module is available
|
||
if (!fs || !fs.promises) {
|
||
console.warn('FileSystemStorage.getStorageStatus: fs module not available, returning default values')
|
||
return {
|
||
type: 'filesystem',
|
||
used: 0,
|
||
quota: null,
|
||
details: {
|
||
nounsCount: 0,
|
||
verbsCount: 0,
|
||
metadataCount: 0,
|
||
directorySizes: {
|
||
nouns: 0,
|
||
verbs: 0,
|
||
metadata: 0,
|
||
index: 0
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
try {
|
||
// Calculate the total size of all files in the storage directories
|
||
let totalSize = 0
|
||
|
||
// Helper function to calculate directory size
|
||
const calculateSize = async (dirPath: string): Promise<number> => {
|
||
let size = 0
|
||
try {
|
||
const files = await fs.promises.readdir(dirPath)
|
||
for (const file of files) {
|
||
const filePath = path.join(dirPath, file)
|
||
const stats = await fs.promises.stat(filePath)
|
||
if (stats.isDirectory()) {
|
||
size += await calculateSize(filePath)
|
||
} else {
|
||
size += stats.size
|
||
}
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(
|
||
`Error calculating size for directory ${dirPath}:`,
|
||
error
|
||
)
|
||
}
|
||
}
|
||
return size
|
||
}
|
||
|
||
// Calculate size for each directory
|
||
const nounsDirSize = await calculateSize(this.nounsDir)
|
||
const verbsDirSize = await calculateSize(this.verbsDir)
|
||
const metadataDirSize = await calculateSize(this.metadataDir)
|
||
const indexDirSize = await calculateSize(this.indexDir)
|
||
|
||
totalSize = nounsDirSize + verbsDirSize + metadataDirSize + indexDirSize
|
||
|
||
// CRITICAL FIX: Use persisted counts instead of directory reads
|
||
// This is O(1) instead of O(n), and handles sharded structure correctly
|
||
const nounsCount = this.totalNounCount
|
||
const verbsCount = this.totalVerbCount
|
||
|
||
// Count metadata files (these are NOT sharded)
|
||
const metadataCount = (
|
||
await fs.promises.readdir(this.metadataDir)
|
||
).filter((file: string) => file.endsWith('.json')).length
|
||
|
||
// Use persisted entity counts by type (O(1) instead of scanning all files)
|
||
const nounTypeCounts: Record<string, number> = Object.fromEntries(this.entityCounts)
|
||
|
||
// Skip the expensive metadata file scan since we have counts
|
||
const metadataFiles: string[] = [] // Empty array to skip the loop below
|
||
for (const file of metadataFiles) {
|
||
if (file.endsWith('.json')) {
|
||
try {
|
||
const filePath = path.join(this.metadataDir, file)
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const metadata = JSON.parse(data)
|
||
if (metadata.noun) {
|
||
nounTypeCounts[metadata.noun] =
|
||
(nounTypeCounts[metadata.noun] || 0) + 1
|
||
}
|
||
} catch (error) {
|
||
console.error(`Error reading metadata file ${file}:`, error)
|
||
}
|
||
}
|
||
}
|
||
|
||
return {
|
||
type: 'filesystem',
|
||
used: totalSize,
|
||
quota: null, // File system doesn't provide quota information
|
||
details: {
|
||
rootDirectory: this.rootDir,
|
||
nounsCount,
|
||
verbsCount,
|
||
metadataCount,
|
||
nounsDirSize,
|
||
verbsDirSize,
|
||
metadataDirSize,
|
||
indexDirSize,
|
||
nounTypes: nounTypeCounts
|
||
}
|
||
}
|
||
} catch (error) {
|
||
console.error('Failed to get storage status:', error)
|
||
return {
|
||
type: 'filesystem',
|
||
used: 0,
|
||
quota: null,
|
||
details: { error: String(error) }
|
||
}
|
||
}
|
||
}
|
||
|
||
// Removed 10 *_internal method overrides - now inherit from BaseStorage's type-first implementation
|
||
// Removed 2 pagination methods (getNounsWithPagination, getVerbsWithPagination) - use BaseStorage's implementation
|
||
|
||
public override supportsMultiProcessLocking(): boolean {
|
||
return true
|
||
}
|
||
|
||
/**
|
||
* Acquire the process-level writer lock for this data directory. Called by
|
||
* `Brainy.init()` in writer mode. Refuses to open if another live writer
|
||
* holds the lock — the worst possible default is to "succeed" with stale
|
||
* indexes and silently lie about query results.
|
||
*
|
||
* Lock file: `<rootDir>/locks/_writer.lock`.
|
||
*
|
||
* Stale-lock detection: a lock is considered stale when ALL of:
|
||
* 1. Same hostname as the current process (cross-host PID checks are unsafe).
|
||
* 2. The recorded PID is no longer alive (`process.kill(pid, 0)` → ESRCH), OR
|
||
* `lastHeartbeat` is older than `WRITER_STALE_THRESHOLD_MS` (60s).
|
||
* Stale locks are overwritten with a warning. `force: true` overrides even live locks.
|
||
*/
|
||
public override async acquireWriterLock(options?: { force?: boolean }): Promise<WriterLockInfo | null> {
|
||
await this.ensureInitialized()
|
||
await this.ensureDirectoryExists(this.lockDir)
|
||
|
||
const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE)
|
||
const os = await import('node:os')
|
||
const hostname = os.hostname()
|
||
const now = new Date().toISOString()
|
||
|
||
const existing = await this.readWriterLock()
|
||
if (existing && !options?.force) {
|
||
// Same-process re-open: a second Brainy instance in this Node process
|
||
// (e.g. test "simulate server restart" patterns, or a consumer that
|
||
// explicitly re-instantiates without closing first). This isn't the
|
||
// dangerous cross-process case the lock exists to prevent — the two
|
||
// instances share a memory space and can't silently diverge from each
|
||
// other beyond what their callers already see. Warn and take over the lock.
|
||
if (existing.pid === (typeof process !== 'undefined' ? process.pid : 0) &&
|
||
existing.hostname === hostname) {
|
||
console.warn(
|
||
`[brainy] Re-acquiring writer lock for ${this.rootDir} held by the same process (PID ${existing.pid}). ` +
|
||
`If you intended to keep the previous Brainy instance alive, this is a bug — close it first.`
|
||
)
|
||
} else {
|
||
const stale = await this.isWriterLockStale(existing)
|
||
if (!stale) {
|
||
// Consumer-facing error contract: callers detect this case via
|
||
// err.code and read the holder's details from err.lockInfo.
|
||
const err = new Error(
|
||
`Another writer holds this Brainy directory.\n` +
|
||
` PID: ${existing.pid} on host ${existing.hostname}\n` +
|
||
` Started: ${existing.startedAt}\n` +
|
||
` Heartbeat: ${existing.lastHeartbeat}\n` +
|
||
` Version: ${existing.version}\n` +
|
||
` Directory: ${this.rootDir}\n\n` +
|
||
`For diagnostic queries against this live store, use:\n` +
|
||
` const reader = await Brainy.openReadOnly({ storage: { type: 'filesystem', path: '${this.rootDir}' } })\n\n` +
|
||
`If you have verified the existing lock is stale (e.g. a crashed writer on a different host that PID liveness cannot reach), pass { force: true }.`
|
||
) as Error & { code: string; lockInfo: WriterLockInfo }
|
||
err.code = 'BRAINY_WRITER_LOCKED'
|
||
err.lockInfo = existing
|
||
throw err
|
||
}
|
||
console.warn(
|
||
`[brainy] Overwriting stale writer lock for ${this.rootDir} ` +
|
||
`(PID ${existing.pid} on ${existing.hostname} appears dead).`
|
||
)
|
||
}
|
||
} else if (existing && options?.force) {
|
||
console.warn(
|
||
`[brainy] Force-overwriting writer lock for ${this.rootDir} ` +
|
||
`(was held by PID ${existing.pid} on ${existing.hostname}).`
|
||
)
|
||
}
|
||
|
||
const info: WriterLockInfo = {
|
||
pid: typeof process !== 'undefined' && process.pid ? process.pid : 0,
|
||
hostname,
|
||
startedAt: existing && options?.force ? existing.startedAt : now,
|
||
lastHeartbeat: now,
|
||
version: getBrainyVersion(),
|
||
rootDir: this.rootDir
|
||
}
|
||
|
||
await this.writeFileAtomic(lockFile, JSON.stringify(info, null, 2))
|
||
this.writerLockInfo = info
|
||
|
||
// Heartbeat — rewrite lastHeartbeat every WRITER_HEARTBEAT_MS so other
|
||
// processes can tell a live writer from one that crashed without releasing.
|
||
this.writerLockHeartbeat = setInterval(() => {
|
||
this.refreshWriterLockHeartbeat().catch((err) => {
|
||
console.warn('[brainy] Failed to refresh writer lock heartbeat:', err)
|
||
})
|
||
}, FileSystemStorage.WRITER_HEARTBEAT_MS)
|
||
if (typeof this.writerLockHeartbeat.unref === 'function') {
|
||
// Don't keep the event loop alive just for the heartbeat.
|
||
this.writerLockHeartbeat.unref()
|
||
}
|
||
|
||
return info
|
||
}
|
||
|
||
public override async releaseWriterLock(): Promise<void> {
|
||
if (this.writerLockHeartbeat) {
|
||
clearInterval(this.writerLockHeartbeat)
|
||
this.writerLockHeartbeat = undefined
|
||
}
|
||
if (!this.writerLockInfo) {
|
||
return
|
||
}
|
||
const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE)
|
||
try {
|
||
// Only delete if we still own it — avoid clobbering a successor that
|
||
// claimed the lock via force-override.
|
||
const current = await this.readWriterLock()
|
||
if (current && current.pid === this.writerLockInfo.pid && current.hostname === this.writerLockInfo.hostname) {
|
||
await fs.promises.unlink(lockFile)
|
||
}
|
||
} catch (err: any) {
|
||
if (err.code !== 'ENOENT') {
|
||
console.warn('[brainy] Failed to release writer lock file:', err)
|
||
}
|
||
} finally {
|
||
this.writerLockInfo = undefined
|
||
}
|
||
}
|
||
|
||
public override async readWriterLock(): Promise<WriterLockInfo | null> {
|
||
await this.ensureInitialized()
|
||
const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE)
|
||
try {
|
||
const raw = await fs.promises.readFile(lockFile, 'utf-8')
|
||
return JSON.parse(raw) as WriterLockInfo
|
||
} catch (err: any) {
|
||
if (err.code === 'ENOENT') return null
|
||
console.warn('[brainy] Failed to read writer lock file:', err)
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Atomically refresh `lastHeartbeat` on the writer lock we own. Skips if the
|
||
* lock file has been deleted out from under us (e.g. operator removed it).
|
||
*/
|
||
private async refreshWriterLockHeartbeat(): Promise<void> {
|
||
if (!this.writerLockInfo) return
|
||
const current = await this.readWriterLock()
|
||
if (!current) return
|
||
// Defensive: don't overwrite if a successor claimed the lock.
|
||
if (current.pid !== this.writerLockInfo.pid || current.hostname !== this.writerLockInfo.hostname) {
|
||
return
|
||
}
|
||
const updated: WriterLockInfo = {
|
||
...this.writerLockInfo,
|
||
lastHeartbeat: new Date().toISOString()
|
||
}
|
||
const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE)
|
||
await this.writeFileAtomic(lockFile, JSON.stringify(updated, null, 2))
|
||
this.writerLockInfo = updated
|
||
}
|
||
|
||
/**
|
||
* Determine whether an existing writer lock is stale (safe to overwrite).
|
||
* Same hostname and (dead PID OR heartbeat older than threshold) → stale.
|
||
* Different hostname → cannot prove stale, treat as live.
|
||
*/
|
||
private async isWriterLockStale(lock: WriterLockInfo): Promise<boolean> {
|
||
const os = await import('node:os')
|
||
if (lock.hostname !== os.hostname()) {
|
||
return false
|
||
}
|
||
const heartbeatAge = Date.now() - new Date(lock.lastHeartbeat).getTime()
|
||
const pidAlive = this.isPidAlive(lock.pid)
|
||
if (!pidAlive) return true
|
||
return heartbeatAge > FileSystemStorage.WRITER_STALE_THRESHOLD_MS
|
||
}
|
||
|
||
/**
|
||
* `process.kill(pid, 0)` sends signal 0 — no signal is actually delivered;
|
||
* it just checks whether the kernel still considers `pid` reachable from
|
||
* this process. ESRCH = no such process. EPERM = exists but we can't signal
|
||
* it, which still proves it's alive.
|
||
*/
|
||
private isPidAlive(pid: number): boolean {
|
||
if (!pid || pid <= 0) return false
|
||
try {
|
||
process.kill(pid, 0)
|
||
return true
|
||
} catch (err: any) {
|
||
if (err.code === 'EPERM') return true
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Atomic write via temp-file-then-rename so concurrent readers never see a
|
||
* half-written lock JSON. Reused by writer-lock writes + heartbeat.
|
||
*/
|
||
private async writeFileAtomic(filePath: string, contents: string): Promise<void> {
|
||
const tmp = `${filePath}.tmp-${process.pid}-${Date.now()}`
|
||
await fs.promises.writeFile(tmp, contents)
|
||
await fs.promises.rename(tmp, filePath)
|
||
}
|
||
|
||
/**
|
||
* Start watching for cross-process flush requests. Called by Brainy.init()
|
||
* in writer mode. Polls `locks/_flush_requests/` every
|
||
* FLUSH_WATCH_INTERVAL_MS — each new `.req` file triggers the supplied
|
||
* callback (`brain.flush()`), after which an `.ack` is written to
|
||
* `locks/_flush_responses/` with the same request ID. Stale `.req` files
|
||
* (>FLUSH_REQUEST_TTL_MS) are garbage-collected on every tick.
|
||
*/
|
||
public override startFlushRequestWatcher(onRequest: () => Promise<void>): void {
|
||
if (this.flushWatcherInterval) return // already watching
|
||
this.flushWatcherOnRequest = onRequest
|
||
|
||
const reqDir = path.join(this.lockDir, FileSystemStorage.FLUSH_REQUEST_DIR)
|
||
const ackDir = path.join(this.lockDir, FileSystemStorage.FLUSH_RESPONSE_DIR)
|
||
|
||
// Ensure both dirs exist up front so the first .req drop doesn't race with mkdir.
|
||
this.ensureDirectoryExists(reqDir).catch(() => {})
|
||
this.ensureDirectoryExists(ackDir).catch(() => {})
|
||
|
||
this.flushWatcherInterval = setInterval(() => {
|
||
if (this.flushWatcherInFlight) return // skip overlapping tick
|
||
this.flushWatcherInFlight = true
|
||
this.processFlushRequests(reqDir, ackDir).finally(() => {
|
||
this.flushWatcherInFlight = false
|
||
})
|
||
}, FileSystemStorage.FLUSH_WATCH_INTERVAL_MS)
|
||
if (typeof this.flushWatcherInterval.unref === 'function') {
|
||
this.flushWatcherInterval.unref()
|
||
}
|
||
}
|
||
|
||
public override stopFlushRequestWatcher(): void {
|
||
if (this.flushWatcherInterval) {
|
||
clearInterval(this.flushWatcherInterval)
|
||
this.flushWatcherInterval = undefined
|
||
}
|
||
this.flushWatcherOnRequest = undefined
|
||
}
|
||
|
||
/**
|
||
* Process any pending `.req` files: invoke the flush callback once, then
|
||
* write an `.ack` for each request. Multiple requests that arrived in the
|
||
* same tick share a single flush — they all see the same ack timestamp.
|
||
*/
|
||
private async processFlushRequests(reqDir: string, ackDir: string): Promise<void> {
|
||
let entries: string[]
|
||
try {
|
||
entries = await fs.promises.readdir(reqDir)
|
||
} catch (err: any) {
|
||
if (err.code !== 'ENOENT') {
|
||
console.warn('[brainy] Flush watcher readdir failed:', err)
|
||
}
|
||
return
|
||
}
|
||
|
||
const reqs = entries.filter((e: string) => e.endsWith('.req'))
|
||
if (reqs.length === 0) return
|
||
|
||
// One flush per tick — N pending requests share it.
|
||
const callback = this.flushWatcherOnRequest
|
||
if (!callback) return
|
||
let flushError: Error | null = null
|
||
try {
|
||
await callback()
|
||
} catch (err: any) {
|
||
flushError = err instanceof Error ? err : new Error(String(err))
|
||
console.warn('[brainy] Flush callback threw inside flush-request watcher:', err)
|
||
}
|
||
|
||
const ackTimestamp = new Date().toISOString()
|
||
for (const filename of reqs) {
|
||
const requestId = filename.replace(/\.req$/, '')
|
||
const ackPath = path.join(ackDir, `${requestId}.ack`)
|
||
const ackBody = JSON.stringify({
|
||
requestId,
|
||
completedAt: ackTimestamp,
|
||
ok: !flushError,
|
||
error: flushError ? flushError.message : undefined
|
||
})
|
||
try {
|
||
await this.writeFileAtomic(ackPath, ackBody)
|
||
await fs.promises.unlink(path.join(reqDir, filename))
|
||
} catch (err: any) {
|
||
if (err.code !== 'ENOENT') {
|
||
console.warn('[brainy] Failed to write flush ack:', err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Garbage-collect stale .req files in case a watcher missed them (e.g.
|
||
// the writer was restarted between request and processing).
|
||
const now = Date.now()
|
||
for (const filename of entries) {
|
||
if (!filename.endsWith('.req')) continue
|
||
const fp = path.join(reqDir, filename)
|
||
try {
|
||
const stat = await fs.promises.stat(fp)
|
||
if (now - stat.mtimeMs > FileSystemStorage.FLUSH_REQUEST_TTL_MS) {
|
||
await fs.promises.unlink(fp).catch(() => {})
|
||
}
|
||
} catch {
|
||
// ignore — file already removed
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Inspector side: drop a `.req` file and poll for the corresponding `.ack`.
|
||
* Returns true if the writer acknowledged in time, false on timeout.
|
||
*/
|
||
public override async requestFlushOverFilesystem(timeoutMs: number): Promise<boolean> {
|
||
await this.ensureInitialized()
|
||
const reqDir = path.join(this.lockDir, FileSystemStorage.FLUSH_REQUEST_DIR)
|
||
const ackDir = path.join(this.lockDir, FileSystemStorage.FLUSH_RESPONSE_DIR)
|
||
await this.ensureDirectoryExists(reqDir)
|
||
await this.ensureDirectoryExists(ackDir)
|
||
|
||
const requestId = `${Date.now()}-${process.pid}-${Math.random().toString(36).slice(2, 10)}`
|
||
const reqPath = path.join(reqDir, `${requestId}.req`)
|
||
const ackPath = path.join(ackDir, `${requestId}.ack`)
|
||
const body = JSON.stringify({
|
||
requestId,
|
||
requestedAt: new Date().toISOString(),
|
||
requesterPid: process.pid
|
||
})
|
||
|
||
await this.writeFileAtomic(reqPath, body)
|
||
|
||
const deadline = Date.now() + timeoutMs
|
||
while (Date.now() < deadline) {
|
||
try {
|
||
await fs.promises.access(ackPath, fs.constants.F_OK)
|
||
await fs.promises.unlink(ackPath).catch(() => {})
|
||
return true
|
||
} catch {
|
||
// not yet
|
||
}
|
||
await new Promise((r) => setTimeout(r, FileSystemStorage.FLUSH_POLL_INTERVAL_MS))
|
||
}
|
||
|
||
// Timeout — best effort to clean up our request so the writer doesn't act
|
||
// on stale work later. Watcher will GC anyway after FLUSH_REQUEST_TTL_MS.
|
||
await fs.promises.unlink(reqPath).catch(() => {})
|
||
return false
|
||
}
|
||
|
||
/**
|
||
* Acquire a file-based lock for coordinating operations across multiple processes
|
||
* @param lockKey The key to lock on
|
||
* @param ttl Time to live for the lock in milliseconds (default: 30 seconds)
|
||
* @returns Promise that resolves to true if lock was acquired, false otherwise
|
||
*/
|
||
private async acquireLock(
|
||
lockKey: string,
|
||
ttl: number = 30000
|
||
): Promise<boolean> {
|
||
await this.ensureInitialized()
|
||
|
||
// Ensure lock directory exists
|
||
await this.ensureDirectoryExists(this.lockDir)
|
||
|
||
const lockFile = path.join(this.lockDir, `${lockKey}.lock`)
|
||
const lockValue = `${Date.now()}_${Math.random()}_${process.pid || 'unknown'}`
|
||
const expiresAt = Date.now() + ttl
|
||
|
||
try {
|
||
// Check if lock file already exists and is still valid
|
||
try {
|
||
const lockData = await fs.promises.readFile(lockFile, 'utf-8')
|
||
const lockInfo = JSON.parse(lockData)
|
||
|
||
if (lockInfo.expiresAt > Date.now()) {
|
||
// Lock exists and is still valid
|
||
return false
|
||
}
|
||
} catch (error: any) {
|
||
// If file doesn't exist or can't be read, we can proceed to create the lock
|
||
if (error.code !== 'ENOENT') {
|
||
console.warn(`Error reading lock file ${lockFile}:`, error)
|
||
}
|
||
}
|
||
|
||
// Try to create the lock file
|
||
const lockInfo = {
|
||
lockValue,
|
||
expiresAt,
|
||
pid: process.pid || 'unknown',
|
||
timestamp: Date.now()
|
||
}
|
||
|
||
await fs.promises.writeFile(lockFile, JSON.stringify(lockInfo, null, 2))
|
||
|
||
// Add to active locks for cleanup
|
||
this.activeLocks.add(lockKey)
|
||
|
||
// Schedule automatic cleanup when lock expires
|
||
setTimeout(() => {
|
||
this.releaseLock(lockKey, lockValue).catch((error) => {
|
||
console.warn(`Failed to auto-release expired lock ${lockKey}:`, error)
|
||
})
|
||
}, ttl)
|
||
|
||
return true
|
||
} catch (error) {
|
||
console.warn(`Failed to acquire lock ${lockKey}:`, error)
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Release a file-based lock
|
||
* @param lockKey The key to unlock
|
||
* @param lockValue The value used when acquiring the lock (for verification)
|
||
* @returns Promise that resolves when lock is released
|
||
*/
|
||
private async releaseLock(
|
||
lockKey: string,
|
||
lockValue?: string
|
||
): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const lockFile = path.join(this.lockDir, `${lockKey}.lock`)
|
||
|
||
try {
|
||
// If lockValue is provided, verify it matches before releasing
|
||
if (lockValue) {
|
||
try {
|
||
const lockData = await fs.promises.readFile(lockFile, 'utf-8')
|
||
const lockInfo = JSON.parse(lockData)
|
||
|
||
if (lockInfo.lockValue !== lockValue) {
|
||
// Lock was acquired by someone else, don't release it
|
||
return
|
||
}
|
||
} catch (error: any) {
|
||
// If lock file doesn't exist, that's fine
|
||
if (error.code === 'ENOENT') {
|
||
return
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// Delete the lock file
|
||
await fs.promises.unlink(lockFile)
|
||
|
||
// Remove from active locks
|
||
this.activeLocks.delete(lockKey)
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.warn(`Failed to release lock ${lockKey}:`, error)
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Clean up expired lock files
|
||
*/
|
||
private async cleanupExpiredLocks(): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
try {
|
||
const lockFiles = await fs.promises.readdir(this.lockDir)
|
||
const now = Date.now()
|
||
|
||
for (const lockFile of lockFiles) {
|
||
if (!lockFile.endsWith('.lock')) continue
|
||
|
||
const lockPath = path.join(this.lockDir, lockFile)
|
||
try {
|
||
const lockData = await fs.promises.readFile(lockPath, 'utf-8')
|
||
const lockInfo = JSON.parse(lockData)
|
||
|
||
if (lockInfo.expiresAt <= now) {
|
||
await fs.promises.unlink(lockPath)
|
||
const lockKey = lockFile.replace('.lock', '')
|
||
this.activeLocks.delete(lockKey)
|
||
}
|
||
} catch (error) {
|
||
// If we can't read or parse the lock file, remove it
|
||
try {
|
||
await fs.promises.unlink(lockPath)
|
||
} catch (unlinkError) {
|
||
console.warn(
|
||
`Failed to cleanup invalid lock file ${lockPath}:`,
|
||
unlinkError
|
||
)
|
||
}
|
||
}
|
||
}
|
||
} catch (error) {
|
||
console.warn('Failed to cleanup expired locks:', error)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Save statistics data to storage with file-based locking
|
||
*/
|
||
protected async saveStatisticsData(
|
||
statistics: StatisticsData
|
||
): Promise<void> {
|
||
const lockKey = 'statistics'
|
||
const lockAcquired = await this.acquireLock(lockKey, 10000) // 10 second timeout
|
||
|
||
if (!lockAcquired) {
|
||
console.warn(
|
||
'Failed to acquire lock for statistics update, proceeding without lock'
|
||
)
|
||
}
|
||
|
||
try {
|
||
// Get existing statistics to merge with new data
|
||
const existingStats = await this.getStatisticsData()
|
||
|
||
if (existingStats) {
|
||
// Merge statistics data
|
||
const mergedStats: StatisticsData = {
|
||
totalNodes: Math.max(
|
||
statistics.totalNodes || 0,
|
||
existingStats.totalNodes || 0
|
||
),
|
||
totalEdges: Math.max(
|
||
statistics.totalEdges || 0,
|
||
existingStats.totalEdges || 0
|
||
),
|
||
totalMetadata: Math.max(
|
||
statistics.totalMetadata || 0,
|
||
existingStats.totalMetadata || 0
|
||
),
|
||
// Preserve any additional fields from existing stats
|
||
...existingStats,
|
||
// Override with new values where provided
|
||
...statistics,
|
||
// Always update lastUpdated to current time
|
||
lastUpdated: new Date().toISOString()
|
||
}
|
||
await this.saveStatisticsWithBackwardCompat(mergedStats)
|
||
} else {
|
||
// No existing statistics, save new ones
|
||
const newStats: StatisticsData = {
|
||
...statistics,
|
||
lastUpdated: new Date().toISOString()
|
||
}
|
||
await this.saveStatisticsWithBackwardCompat(newStats)
|
||
}
|
||
} finally {
|
||
if (lockAcquired) {
|
||
await this.releaseLock(lockKey)
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get statistics data from storage
|
||
*/
|
||
protected async getStatisticsData(): Promise<StatisticsData | null> {
|
||
try {
|
||
const statsPath = path.join(this.systemDir, `${STATISTICS_KEY}.json`)
|
||
const data = await fs.promises.readFile(statsPath, 'utf-8')
|
||
return JSON.parse(data)
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error('Error reading statistics:', error)
|
||
}
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Save statistics to storage
|
||
*/
|
||
private async saveStatisticsWithBackwardCompat(statistics: StatisticsData): Promise<void> {
|
||
const statsPath = path.join(this.systemDir, `${STATISTICS_KEY}.json`)
|
||
await this.ensureDirectoryExists(this.systemDir)
|
||
await fs.promises.writeFile(statsPath, JSON.stringify(statistics, null, 2))
|
||
}
|
||
|
||
// =============================================
|
||
// Count Management for O(1) Scalability
|
||
// =============================================
|
||
|
||
/**
|
||
* Initialize counts from filesystem storage
|
||
*/
|
||
protected async initializeCounts(): Promise<void> {
|
||
if (!this.countsFilePath) return
|
||
|
||
try {
|
||
if (await this.fileExists(this.countsFilePath)) {
|
||
const data = await fs.promises.readFile(this.countsFilePath, 'utf-8')
|
||
const counts = JSON.parse(data)
|
||
|
||
// Restore entity counts
|
||
this.entityCounts = new Map(Object.entries(counts.entityCounts || {}))
|
||
this.verbCounts = new Map(Object.entries(counts.verbCounts || {}))
|
||
this.totalNounCount = counts.totalNounCount || 0
|
||
this.totalVerbCount = counts.totalVerbCount || 0
|
||
|
||
// Also populate the cache for backward compatibility
|
||
this.countCache.set('nouns_count', {
|
||
count: this.totalNounCount,
|
||
timestamp: Date.now()
|
||
})
|
||
this.countCache.set('verbs_count', {
|
||
count: this.totalVerbCount,
|
||
timestamp: Date.now()
|
||
})
|
||
} else {
|
||
// If no counts file exists, do one initial count
|
||
await this.initializeCountsFromDisk()
|
||
}
|
||
} catch (error) {
|
||
console.warn('Could not load persisted counts, will initialize from disk:', error)
|
||
await this.initializeCountsFromDisk()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Initialize counts by scanning disk (only done once)
|
||
*/
|
||
private async initializeCountsFromDisk(): Promise<void> {
|
||
try {
|
||
// Count the CANONICAL 8.0 layout (`entities/<kind>/<shard>/<id>/…`) —
|
||
// the tree saveNoun/getNouns actually read and write. The previous scan
|
||
// counted the vestigial `entities/*/hnsw` directories, which the 8.0
|
||
// write path never populates, so a store recovering from a lost or
|
||
// corrupted counts.json re-initialized every counter to ZERO on real
|
||
// data (wrong stats/boot logs and a mis-sized rebuild-strategy
|
||
// decision at open).
|
||
const nouns = await this.scanCanonicalEntities('nouns')
|
||
this.totalNounCount = nouns.count
|
||
const verbs = await this.scanCanonicalEntities('verbs')
|
||
this.totalVerbCount = verbs.count
|
||
|
||
// Sample some entities for the type distribution (don't read all).
|
||
// Read the metadata files DIRECTLY with fs — this runs inside init(),
|
||
// and every guarded accessor (getNounMetadata → ensureInitialized)
|
||
// re-enters init() from here, deadlocking the open. (Latent in the old
|
||
// code too: its scan of the empty hnsw dirs just never sampled.)
|
||
for (const entityDir of nouns.sampleDirs) {
|
||
const metadata = await this.readEntityMetadataRaw(entityDir)
|
||
if (metadata) {
|
||
const type = metadata.noun || 'default'
|
||
this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1)
|
||
}
|
||
}
|
||
|
||
// Extrapolate counts if we sampled
|
||
const sampleSize = nouns.sampleDirs.length
|
||
if (sampleSize < this.totalNounCount && sampleSize > 0) {
|
||
const multiplier = this.totalNounCount / sampleSize
|
||
for (const [type, count] of this.entityCounts.entries()) {
|
||
this.entityCounts.set(type, Math.round(count * multiplier))
|
||
}
|
||
}
|
||
|
||
await this.persistCounts()
|
||
} catch (error) {
|
||
console.error('Error initializing counts from disk:', error)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Walk the canonical `entities/<kind>/<2-hex-shard>/<id>/` tree, counting
|
||
* one entity per id directory (the layout `getNounVectorPath`/`getNouns`
|
||
* use). Returns up to 100 sampled entity directories (absolute paths) —
|
||
* nouns feed the type-distribution estimate above. An absent tree (fresh
|
||
* store) counts zero.
|
||
*/
|
||
private async scanCanonicalEntities(
|
||
kind: 'nouns' | 'verbs'
|
||
): Promise<{ count: number; sampleDirs: string[] }> {
|
||
const base = path.join(this.rootDir, 'entities', kind)
|
||
const SAMPLE_MAX = 100
|
||
let count = 0
|
||
const sampleDirs: string[] = []
|
||
try {
|
||
const shards = await fs.promises.readdir(base, { withFileTypes: true })
|
||
for (const shard of shards) {
|
||
if (!shard.isDirectory() || !/^[0-9a-f]{2}$/i.test(shard.name)) continue
|
||
const shardPath = path.join(base, shard.name)
|
||
const ids = await fs.promises.readdir(shardPath, { withFileTypes: true })
|
||
for (const entry of ids) {
|
||
if (!entry.isDirectory()) continue
|
||
count++
|
||
if (sampleDirs.length < SAMPLE_MAX) {
|
||
sampleDirs.push(path.join(shardPath, entry.name))
|
||
}
|
||
}
|
||
}
|
||
} catch (error: any) {
|
||
if (error?.code !== 'ENOENT') throw error
|
||
}
|
||
return { count, sampleDirs }
|
||
}
|
||
|
||
/**
|
||
* Read one canonical entity's `metadata.json` (or `.json.gz`) directly with
|
||
* fs — NO guarded accessors. Used only by the init-time count recovery,
|
||
* where `getNounMetadata`'s `ensureInitialized()` would re-enter `init()`.
|
||
* @param entityDir - Absolute `entities/<kind>/<shard>/<id>` directory.
|
||
* @returns The parsed metadata, or null when absent/unreadable.
|
||
*/
|
||
private async readEntityMetadataRaw(entityDir: string): Promise<any | null> {
|
||
const base = path.join(entityDir, 'metadata.json')
|
||
try {
|
||
return JSON.parse(await fs.promises.readFile(base, 'utf-8'))
|
||
} catch {
|
||
// fall through to the compressed variant
|
||
}
|
||
try {
|
||
const gz = await fs.promises.readFile(`${base}.gz`)
|
||
return JSON.parse(zlib.gunzipSync(gz).toString('utf-8'))
|
||
} catch {
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Persist counts to filesystem storage
|
||
*/
|
||
protected async persistCounts(): Promise<void> {
|
||
if (!this.countsFilePath) return
|
||
|
||
try {
|
||
const counts = {
|
||
entityCounts: Object.fromEntries(this.entityCounts),
|
||
verbCounts: Object.fromEntries(this.verbCounts),
|
||
totalNounCount: this.totalNounCount,
|
||
totalVerbCount: this.totalVerbCount,
|
||
lastUpdated: new Date().toISOString()
|
||
}
|
||
|
||
await fs.promises.writeFile(
|
||
this.countsFilePath,
|
||
JSON.stringify(counts, null, 2)
|
||
)
|
||
} catch (error) {
|
||
console.error('Error persisting counts:', error)
|
||
}
|
||
}
|
||
|
||
|
||
|
||
// =============================================
|
||
// Intelligent Directory Sharding
|
||
// =============================================
|
||
|
||
|
||
|
||
|
||
|
||
/**
|
||
* Migrate a single file atomically
|
||
*/
|
||
private async migrateFile(
|
||
fileInfo: { oldPath: string; id: string; type: 'noun' | 'verb' },
|
||
fromDepth: number,
|
||
toDepth: number
|
||
): Promise<void> {
|
||
const baseDir = fileInfo.type === 'noun' ? this.nounsDir : this.verbsDir
|
||
|
||
// Calculate old path (already known)
|
||
const oldPath = fileInfo.oldPath
|
||
|
||
// Calculate new path using target depth
|
||
const shard = fileInfo.id.substring(0, 2).toLowerCase()
|
||
const newPath = path.join(baseDir, shard, `${fileInfo.id}.json`)
|
||
|
||
// Check if file already exists at new location
|
||
if (await this.fileExists(newPath)) {
|
||
// File already migrated or duplicate - skip
|
||
return
|
||
}
|
||
|
||
// Atomic rename/move
|
||
await fs.promises.rename(oldPath, newPath)
|
||
}
|
||
|
||
|
||
|
||
|
||
/**
|
||
* Whether this store already holds canonical 8.0 entities.
|
||
*
|
||
* 8.0 writes nouns to `entities/nouns/<shard>/<id>/vectors.json` (see
|
||
* `getNounVectorPath`). This checks that canonical shard tree — the one the
|
||
* DB actually reads and writes (the same `entities/nouns/<hex>` shards
|
||
* `getNounsWithPagination` walks) — so the new-vs-established boot log is
|
||
* truthful. (The 7.x hnsw sharding probe this replaced inspected a directory
|
||
* the 8.0 write path never populated, so it mislabeled every established
|
||
* store "New installation" on every boot; that probe and its depth-migration
|
||
* machinery are removed.)
|
||
*
|
||
* @returns true if at least one 2-hex shard directory (00–ff) exists under
|
||
* `entities/nouns/`, i.e. the store has previously persisted entities.
|
||
*/
|
||
private async hasCanonicalEntities(): Promise<boolean> {
|
||
const canonicalNounsDir = path.join(this.rootDir, 'entities', 'nouns')
|
||
try {
|
||
const entries = await fs.promises.readdir(canonicalNounsDir, {
|
||
withFileTypes: true
|
||
})
|
||
// A populated store has ≥1 two-hex shard dir (00–ff). The vestigial
|
||
// `hnsw` subdir is 4 chars and correctly excluded by the hex test.
|
||
return entries.some(
|
||
(e: any) => e.isDirectory() && /^[0-9a-f]{2}$/i.test(e.name)
|
||
)
|
||
} catch {
|
||
// Directory absent (fresh store) or unreadable → no persisted entities.
|
||
return false
|
||
}
|
||
}
|
||
|
||
|
||
/**
|
||
* Check if a file exists (handles both sharded and non-sharded)
|
||
*/
|
||
private async fileExists(filePath: string): Promise<boolean> {
|
||
try {
|
||
await fs.promises.access(filePath, fs.constants.F_OK)
|
||
return true
|
||
} catch {
|
||
return false
|
||
}
|
||
}
|
||
|
||
// =============================================
|
||
// HNSW Index Persistence
|
||
// =============================================
|
||
|
||
/**
|
||
* Get vector for a noun
|
||
* Uses BaseStorage's getNoun (type-first paths)
|
||
*/
|
||
public async getNounVector(id: string): Promise<number[] | null> {
|
||
const noun = await this.getNoun(id)
|
||
return noun ? noun.vector : null
|
||
}
|
||
|
||
/**
|
||
* Save HNSW graph data for a noun
|
||
*
|
||
* Uses BaseStorage's getNoun/saveNoun (type-first paths)
|
||
* CRITICAL: Preserves mutex locking to prevent read-modify-write races
|
||
*/
|
||
public async saveVectorIndexData(nounId: string, hnswData: {
|
||
level: number
|
||
connections: Record<string, string[]>
|
||
}): Promise<void> {
|
||
const lockKey = `hnsw/${nounId}`
|
||
|
||
// CRITICAL FIX: Mutex lock to prevent read-modify-write races
|
||
// Problem: Without mutex, concurrent operations can:
|
||
// 1. Thread A reads noun (connections: [1,2,3])
|
||
// 2. Thread B reads noun (connections: [1,2,3])
|
||
// 3. Thread A adds connection 4, writes [1,2,3,4]
|
||
// 4. Thread B adds connection 5, writes [1,2,3,5] ← Connection 4 LOST!
|
||
// Solution: Mutex serializes operations per entity (like Memory/OPFS adapters)
|
||
// Production scale: Prevents corruption at 1000+ concurrent operations
|
||
|
||
// Wait for any pending operations on this entity
|
||
while (this.hnswLocks.has(lockKey)) {
|
||
await this.hnswLocks.get(lockKey)
|
||
}
|
||
|
||
// Acquire lock
|
||
let releaseLock!: () => void
|
||
const lockPromise = new Promise<void>(resolve => { releaseLock = resolve })
|
||
this.hnswLocks.set(lockKey, lockPromise)
|
||
|
||
try {
|
||
// Use BaseStorage's getNoun (type-first paths)
|
||
// Read existing noun data (if exists)
|
||
const existingNoun = await this.getNoun(nounId)
|
||
|
||
if (!existingNoun) {
|
||
// Noun doesn't exist - cannot update HNSW data for non-existent noun
|
||
throw new Error(`Cannot save HNSW data: noun ${nounId} not found`)
|
||
}
|
||
|
||
// Convert connections from Record to Map format for storage
|
||
const connectionsMap = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(hnswData.connections)) {
|
||
connectionsMap.set(Number(level), new Set(nodeIds))
|
||
}
|
||
|
||
// Preserve id and vector, update only HNSW graph metadata
|
||
const updatedNoun: HNSWNoun = {
|
||
...existingNoun,
|
||
level: hnswData.level,
|
||
connections: connectionsMap
|
||
}
|
||
|
||
// Use BaseStorage's saveNoun (type-first paths, atomic write)
|
||
await this.saveNoun(updatedNoun)
|
||
} finally {
|
||
// Release lock (ALWAYS runs, even if error thrown)
|
||
this.hnswLocks.delete(lockKey)
|
||
releaseLock()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get HNSW graph data for a noun
|
||
* Uses BaseStorage's getNoun (type-first paths)
|
||
*/
|
||
public async getVectorIndexData(nounId: string): Promise<{
|
||
level: number
|
||
connections: Record<string, string[]>
|
||
} | null> {
|
||
const noun = await this.getNoun(nounId)
|
||
|
||
if (!noun) {
|
||
return null
|
||
}
|
||
|
||
// Convert connections from Map to Record format
|
||
const connectionsRecord: Record<string, string[]> = {}
|
||
if (noun.connections) {
|
||
for (const [level, nodeIds] of noun.connections.entries()) {
|
||
connectionsRecord[String(level)] = Array.from(nodeIds)
|
||
}
|
||
}
|
||
|
||
return {
|
||
level: noun.level || 0,
|
||
connections: connectionsRecord
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Save HNSW system data (entry point, max level)
|
||
*
|
||
* CRITICAL FIX: Mutex lock + atomic write to prevent race conditions
|
||
*/
|
||
public async saveHNSWSystem(systemData: {
|
||
entryPointId: string | null
|
||
maxLevel: number
|
||
}): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const lockKey = 'hnsw/system'
|
||
|
||
// CRITICAL FIX: Mutex lock to serialize system updates
|
||
// System data (entry point, max level) updated frequently during HNSW construction
|
||
// Without mutex, concurrent updates can lose data (same as entity-level problem)
|
||
|
||
// Wait for any pending system updates
|
||
while (this.hnswLocks.has(lockKey)) {
|
||
await this.hnswLocks.get(lockKey)
|
||
}
|
||
|
||
// Acquire lock
|
||
let releaseLock!: () => void
|
||
const lockPromise = new Promise<void>(resolve => { releaseLock = resolve })
|
||
this.hnswLocks.set(lockKey, lockPromise)
|
||
|
||
try {
|
||
const filePath = path.join(this.systemDir, 'hnsw-system.json')
|
||
const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}`
|
||
|
||
try {
|
||
// Write to temp file
|
||
await this.ensureDirectoryExists(path.dirname(tempPath))
|
||
await fs.promises.writeFile(tempPath, JSON.stringify(systemData, null, 2))
|
||
|
||
// Atomic rename temp → final (POSIX atomicity guarantee)
|
||
await fs.promises.rename(tempPath, filePath)
|
||
} catch (error: any) {
|
||
// Clean up temp file on any error
|
||
try {
|
||
await fs.promises.unlink(tempPath)
|
||
} catch (cleanupError) {
|
||
// Ignore cleanup errors
|
||
}
|
||
throw error
|
||
}
|
||
} finally {
|
||
// Release lock
|
||
this.hnswLocks.delete(lockKey)
|
||
releaseLock()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get HNSW system data
|
||
*/
|
||
public async getHNSWSystem(): Promise<{
|
||
entryPointId: string | null
|
||
maxLevel: number
|
||
} | null> {
|
||
await this.ensureInitialized()
|
||
|
||
const filePath = path.join(this.systemDir, 'hnsw-system.json')
|
||
|
||
try {
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
return JSON.parse(data)
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error('Error reading HNSW system data:', error)
|
||
}
|
||
return null
|
||
}
|
||
}
|
||
}
|