David Snelling
550bd4a19c
FileSystemStorage.saveBinaryBlob used a bare ${filePath}.tmp suffix for its
atomic-write temp file. Two concurrent same-key calls computed the SAME temp
path; both writeFile'd, the first rename succeeded, the second rename fired
against a missing temp and threw ENOENT. The throw propagated up through
brain.flush() and broke any downstream job that called it.
Reproduced in production (Brainy 7.30 + Cortex 2.7 + mmap-filesystem):
[job-queue] gcs-backup: failed - ENOENT: no such file or directory,
rename '/data/brainy-data/.../_column_index/owner/DELETED.bin.tmp'
-> '/data/brainy-data/.../_column_index/owner/DELETED.bin'
The race was two cron jobs (gcs-backup hourly + flush-brain every 15min) both
calling flush(), triggering column-store compaction over the same fields,
overlapping at the rename. Off-site backups had been failing continuously for
~48 hours.
PATCH
src/storage/adapters/fileSystemStorage.ts:992-999 — saveBinaryBlob now uses a
unique per-writer temp suffix matching the pattern at every other atomic-write
site in the same file (lines 336, 551, 744, 781, 1529, 2908):
const tmpPath = `${filePath}.tmp.${process.pid}.${Date.now()}.${Math.random().toString(36).slice(2)}`
Adds defensive ENOENT swallow on rename: if the temp is gone, the work has
already landed (saveBinaryBlob is idempotent for a given key — all callers
persist the same logical bytes per key). Cleans up the temp on any other
rename failure to avoid orphan .tmp.* files.
SCOPE AUDIT
One bug site. Audit results:
- FileSystemStorage: six sibling atomic-write sites already used unique
suffixes (lines 336/551/744/781/1529/2908); only saveBinaryBlob was the
outlier. All six rechecked. Clean.
- OPFSStorage: WritableStream (no tmp+rename). Not affected.
- GCSStorage / R2Storage / AzureBlobStorage / S3CompatibleStorage: object-store
PUT (atomic at the API). Not affected.
- MemoryStorage: in-memory. Not affected.
- HistoricalStorageAdapter: read-only. Not affected.
- COW / versioning / snapshot / HNSW / aggregation: all delegate to storage
adapters via saveBinaryBlob / writeObjectToPath. They get the fix
automatically by using the patched primitive.
The bare-`.tmp` pattern is now gone repo-wide.
BENEFICIARIES
Beyond the reported column-store-compaction race:
- HNSW connection persistence (src/hnsw/hnswIndex.ts:252 → saveBinaryBlob)
was structurally susceptible to the same race. No production reports of
HNSW failures (probably because HNSW writes are more naturally serialized
by the index lock), but the fix removes the latent issue.
- Any future caller of saveBinaryBlob inherits the safer semantics.
TESTS
New tests/integration/savebinaryblob-concurrent-rename.test.ts (4 tests):
- 20 concurrent saveBinaryBlob(sameKey, ...) all resolve without throwing
- No orphan .tmp.* siblings remain in the blob directory afterwards
- Production-shape: two concurrent compactor passes over 10 column-index
fields (owner, path, permissions, vfsType, modified, createdAt, accessed,
updatedAt, mimeType, size). All 20 calls succeed; each field ends with
valid bytes.
- Single-writer path still produces correct bytes (no regression on common
case).
Verified the first three tests reproduce the production ENOENT error
verbatim on the pre-7.31.1 code path (stashed the fix, watched them fail
with the exact production error).
VERIFICATION
- npx tsc --noEmit: clean
- npm test: 1468 / 1468 unit
- New integration suite: 4/4
- npm run build: clean
CORTEX COMPATIBILITY
Zero changes. The Cortex mmap-filesystem adapter wraps FileSystemStorage
and inherits the patch automatically.
FORWARD-COMPAT
8.0's Db.persist() will route through the same patched primitive; no
additional work needed when the immutable Db API ships.
2973 lines
101 KiB
TypeScript
2973 lines
101 KiB
TypeScript
/**
|
||
* File System Storage Adapter
|
||
* File system storage adapter for Node.js environments
|
||
*/
|
||
|
||
import {
|
||
GraphVerb,
|
||
HNSWNoun,
|
||
HNSWVerb,
|
||
NounMetadata,
|
||
VerbMetadata,
|
||
HNSWNounWithMetadata,
|
||
HNSWVerbWithMetadata,
|
||
StatisticsData,
|
||
NounType
|
||
} from '../../coreTypes.js'
|
||
import {
|
||
BaseStorage,
|
||
StorageBatchConfig,
|
||
SYSTEM_DIR,
|
||
STATISTICS_KEY,
|
||
WriterLockInfo
|
||
} from '../baseStorage.js'
|
||
import { getBrainyVersion } from '../../utils/index.js'
|
||
|
||
// Type aliases for better readability
|
||
type HNSWNode = HNSWNoun
|
||
type Edge = HNSWVerb
|
||
|
||
// 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
|
||
private readonly MAX_SHARDS = 256 // Hex range: 00-ff
|
||
private cachedShardingDepth: number = this.SHARDING_DEPTH // Always use fixed depth
|
||
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)
|
||
|
||
/**
|
||
* 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 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 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)
|
||
|
||
// 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()
|
||
|
||
// Detect existing sharding structure and migrate if needed
|
||
const detectedDepth = await this.detectExistingShardingDepth()
|
||
|
||
if (detectedDepth !== null && detectedDepth !== this.SHARDING_DEPTH) {
|
||
// Migration needed: existing structure doesn't match our fixed depth
|
||
console.log(`📦 Brainy Storage Migration`)
|
||
console.log(` Current structure: depth ${detectedDepth}`)
|
||
console.log(` Target structure: depth ${this.SHARDING_DEPTH}`)
|
||
console.log(` Entities to migrate: ${this.totalNounCount}`)
|
||
|
||
await this.migrateShardingStructure(detectedDepth, this.SHARDING_DEPTH)
|
||
|
||
console.log(`✅ Migration complete - now using depth ${this.SHARDING_DEPTH} sharding`)
|
||
} else if (detectedDepth === null) {
|
||
// New installation
|
||
console.log(`📁 New installation: using depth ${this.SHARDING_DEPTH} sharding (optimal for 1-2.5M entities)`)
|
||
} else {
|
||
// Already using correct depth
|
||
console.log(`📁 Using depth ${this.SHARDING_DEPTH} sharding (${this.totalNounCount} entities)`)
|
||
}
|
||
|
||
// Always use fixed depth after migration/detection
|
||
this.cachedShardingDepth = this.SHARDING_DEPTH
|
||
|
||
// 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
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Save a node to storage
|
||
* CRITICAL FIX: Added atomic write pattern to prevent file corruption during concurrent imports
|
||
*/
|
||
protected async saveNode(node: HNSWNode): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Convert connections Map to a serializable format
|
||
// CRITICAL: Only save lightweight vector data (no metadata)
|
||
// Metadata is saved separately via saveNounMetadata() (2-file system)
|
||
const serializableNode = {
|
||
id: node.id,
|
||
vector: node.vector,
|
||
connections: this.mapToObject(node.connections, (set) =>
|
||
Array.from(set as Set<string>)
|
||
),
|
||
level: node.level || 0
|
||
// NO metadata field - saved separately for scalability
|
||
}
|
||
|
||
const filePath = this.getNodePath(node.id)
|
||
const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}`
|
||
|
||
try {
|
||
// ATOMIC WRITE SEQUENCE:
|
||
// 1. Write to temp file
|
||
await this.ensureDirectoryExists(path.dirname(tempPath))
|
||
await fs.promises.writeFile(tempPath, JSON.stringify(serializableNode, null, 2))
|
||
|
||
// 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes)
|
||
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
|
||
}
|
||
|
||
// Count tracking happens in baseStorage.saveNounMetadata_internal
|
||
// This fixes the race condition where metadata didn't exist yet
|
||
}
|
||
|
||
/**
|
||
* Get a node from storage
|
||
*/
|
||
protected async getNode(id: string): Promise<HNSWNode | null> {
|
||
await this.ensureInitialized()
|
||
|
||
// Clean, predictable path - no backward compatibility needed
|
||
const filePath = this.getNodePath(id)
|
||
try {
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const parsedNode = JSON.parse(data)
|
||
|
||
// Convert serialized connections back to Map<number, Set<string>>
|
||
const connections = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(parsedNode.connections)) {
|
||
connections.set(Number(level), new Set(nodeIds as string[]))
|
||
}
|
||
|
||
// CRITICAL: Only return lightweight vector data (no metadata)
|
||
// Metadata is retrieved separately via getNounMetadata() (2-file system)
|
||
return {
|
||
id: parsedNode.id,
|
||
vector: parsedNode.vector,
|
||
connections,
|
||
level: parsedNode.level || 0
|
||
// NO metadata field - retrieved separately for scalability
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error reading node ${id}:`, error)
|
||
}
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get all nodes from storage
|
||
* CRITICAL FIX: Now scans sharded subdirectories (depth=1)
|
||
* Previously only scanned flat directory, causing rebuild to find 0 entities
|
||
*/
|
||
protected async getAllNodes(): Promise<HNSWNode[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const allNodes: HNSWNode[] = []
|
||
try {
|
||
// FIX: Use sharded file discovery instead of flat directory read
|
||
// This scans all 256 shard subdirectories (00-ff) to find actual files
|
||
const files = await this.getAllShardedFiles(this.nounsDir)
|
||
|
||
for (const file of files) {
|
||
// Extract ID from filename and use sharded path
|
||
const id = file.replace('.json', '')
|
||
const filePath = this.getNodePath(id)
|
||
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const parsedNode = JSON.parse(data)
|
||
|
||
// Convert serialized connections back to Map<number, Set<string>>
|
||
const connections = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(
|
||
parsedNode.connections
|
||
)) {
|
||
connections.set(Number(level), new Set(nodeIds as string[]))
|
||
}
|
||
|
||
allNodes.push({
|
||
id: parsedNode.id,
|
||
vector: parsedNode.vector,
|
||
connections,
|
||
level: parsedNode.level || 0
|
||
})
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error reading directory ${this.nounsDir}:`, error)
|
||
}
|
||
}
|
||
return allNodes
|
||
}
|
||
|
||
/**
|
||
* Get nodes by noun type
|
||
* CRITICAL FIX: Now scans sharded subdirectories (depth=1)
|
||
* @param nounType The noun type to filter by
|
||
* @returns Promise that resolves to an array of nodes of the specified noun type
|
||
*/
|
||
protected async getNodesByNounType(nounType: string): Promise<HNSWNode[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const nouns: HNSWNode[] = []
|
||
try {
|
||
// FIX: Use sharded file discovery instead of flat directory read
|
||
const files = await this.getAllShardedFiles(this.nounsDir)
|
||
|
||
for (const file of files) {
|
||
// Extract ID from filename and use sharded path
|
||
const nodeId = file.replace('.json', '')
|
||
const filePath = this.getNodePath(nodeId)
|
||
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const parsedNode = JSON.parse(data)
|
||
|
||
// Filter by noun type using metadata
|
||
const metadata = await this.getMetadata(nodeId)
|
||
if (metadata && metadata.noun === nounType) {
|
||
// Convert serialized connections back to Map<number, Set<string>>
|
||
const connections = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(
|
||
parsedNode.connections
|
||
)) {
|
||
connections.set(Number(level), new Set(nodeIds as string[]))
|
||
}
|
||
|
||
nouns.push({
|
||
id: parsedNode.id,
|
||
vector: parsedNode.vector,
|
||
connections,
|
||
level: parsedNode.level || 0
|
||
})
|
||
}
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error reading directory ${this.nounsDir}:`, error)
|
||
}
|
||
}
|
||
|
||
return nouns
|
||
}
|
||
|
||
/**
|
||
* Delete a node from storage
|
||
*/
|
||
protected async deleteNode(id: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const filePath = this.getNodePath(id)
|
||
|
||
// Load metadata to get type for count update (separate storage)
|
||
try {
|
||
const metadata = await this.getNounMetadata(id)
|
||
if (metadata) {
|
||
const type = metadata.noun || 'default'
|
||
this.decrementEntityCount(type)
|
||
}
|
||
} catch {
|
||
// Metadata might not exist, that's ok
|
||
}
|
||
|
||
try {
|
||
await fs.promises.unlink(filePath)
|
||
|
||
// Persist counts periodically
|
||
if (this.totalNounCount % 10 === 0) {
|
||
await this.persistCounts()
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error deleting node file ${filePath}:`, error)
|
||
throw error
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Save an edge to storage
|
||
* CRITICAL FIX: Added atomic write pattern to prevent file corruption during concurrent imports
|
||
*/
|
||
protected async saveEdge(edge: Edge): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Convert connections Map to a serializable format
|
||
// ARCHITECTURAL FIX: Include core relational fields in verb vector file
|
||
// These fields are essential for 90% of operations - no metadata lookup needed
|
||
const serializableEdge = {
|
||
id: edge.id,
|
||
vector: edge.vector,
|
||
connections: this.mapToObject(edge.connections, (set) =>
|
||
Array.from(set as Set<string>)
|
||
),
|
||
|
||
// CORE RELATIONAL DATA
|
||
verb: edge.verb,
|
||
sourceId: edge.sourceId,
|
||
targetId: edge.targetId,
|
||
|
||
// User metadata (if any) - saved separately for scalability
|
||
// metadata field is saved separately via saveVerbMetadata()
|
||
}
|
||
|
||
const filePath = this.getVerbPath(edge.id)
|
||
const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}`
|
||
|
||
try {
|
||
// ATOMIC WRITE SEQUENCE:
|
||
// 1. Write to temp file
|
||
await this.ensureDirectoryExists(path.dirname(tempPath))
|
||
await fs.promises.writeFile(tempPath, JSON.stringify(serializableEdge, null, 2))
|
||
|
||
// 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes)
|
||
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
|
||
}
|
||
|
||
// Count tracking happens in baseStorage.saveVerbMetadata_internal
|
||
// This fixes the race condition where metadata didn't exist yet
|
||
}
|
||
|
||
/**
|
||
* Get an edge from storage
|
||
*/
|
||
protected async getEdge(id: string): Promise<Edge | null> {
|
||
await this.ensureInitialized()
|
||
|
||
const filePath = this.getVerbPath(id)
|
||
try {
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const parsedEdge = JSON.parse(data)
|
||
|
||
// Convert serialized connections back to Map<number, Set<string>>
|
||
const connections = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(parsedEdge.connections)) {
|
||
connections.set(Number(level), new Set(nodeIds as string[]))
|
||
}
|
||
|
||
// Return HNSWVerb with core relational fields (NO metadata field)
|
||
return {
|
||
id: parsedEdge.id,
|
||
vector: parsedEdge.vector,
|
||
connections,
|
||
|
||
// CORE RELATIONAL DATA (read from vector file)
|
||
verb: parsedEdge.verb,
|
||
sourceId: parsedEdge.sourceId,
|
||
targetId: parsedEdge.targetId
|
||
|
||
// ✅ NO metadata field
|
||
// User metadata retrieved separately via getVerbMetadata()
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error reading edge ${id}:`, error)
|
||
}
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get all edges from storage
|
||
* CRITICAL FIX: Now scans sharded subdirectories (depth=1)
|
||
* Previously only scanned flat directory, causing rebuild to find 0 relationships
|
||
*/
|
||
protected async getAllEdges(): Promise<Edge[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const allEdges: Edge[] = []
|
||
try {
|
||
// FIX: Use sharded file discovery instead of flat directory read
|
||
// This scans all 256 shard subdirectories (00-ff) to find actual files
|
||
const files = await this.getAllShardedFiles(this.verbsDir)
|
||
|
||
for (const file of files) {
|
||
// Extract ID from filename and use sharded path
|
||
const id = file.replace('.json', '')
|
||
const filePath = this.getVerbPath(id)
|
||
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const parsedEdge = JSON.parse(data)
|
||
|
||
// Convert serialized connections back to Map<number, Set<string>>
|
||
const connections = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(
|
||
parsedEdge.connections
|
||
)) {
|
||
connections.set(Number(level), new Set(nodeIds as string[]))
|
||
}
|
||
|
||
// Include core relational fields (NO metadata field)
|
||
allEdges.push({
|
||
id: parsedEdge.id,
|
||
vector: parsedEdge.vector,
|
||
connections,
|
||
|
||
// CORE RELATIONAL DATA
|
||
verb: parsedEdge.verb,
|
||
sourceId: parsedEdge.sourceId,
|
||
targetId: parsedEdge.targetId
|
||
|
||
// ✅ NO metadata field
|
||
// User metadata retrieved separately via getVerbMetadata()
|
||
})
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error reading directory ${this.verbsDir}:`, error)
|
||
}
|
||
}
|
||
return allEdges
|
||
}
|
||
|
||
/**
|
||
* Get edges by source
|
||
*/
|
||
protected async getEdgesBySource(sourceId: string): Promise<Edge[]> {
|
||
// This method is deprecated and would require loading metadata for each edge
|
||
// For now, return empty array since this is not efficiently implementable with new storage pattern
|
||
console.warn('getEdgesBySource is deprecated and not efficiently supported in new storage pattern')
|
||
return []
|
||
}
|
||
|
||
/**
|
||
* Get edges by target
|
||
*/
|
||
protected async getEdgesByTarget(targetId: string): Promise<Edge[]> {
|
||
// This method is deprecated and would require loading metadata for each edge
|
||
// For now, return empty array since this is not efficiently implementable with new storage pattern
|
||
console.warn('getEdgesByTarget is deprecated and not efficiently supported in new storage pattern')
|
||
return []
|
||
}
|
||
|
||
/**
|
||
* Get edges by type
|
||
*/
|
||
protected async getEdgesByType(type: string): Promise<Edge[]> {
|
||
// This method is deprecated and would require loading metadata for each edge
|
||
// For now, return empty array since this is not efficiently implementable with new storage pattern
|
||
console.warn('getEdgesByType is deprecated and not efficiently supported in new storage pattern')
|
||
return []
|
||
}
|
||
|
||
/**
|
||
* Delete an edge from storage
|
||
*/
|
||
protected async deleteEdge(id: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Delete the HNSWVerb file using sharded path
|
||
const filePath = this.getVerbPath(id)
|
||
try {
|
||
await fs.promises.unlink(filePath)
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
console.error(`Error deleting edge file ${filePath}:`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// CRITICAL: Also delete verb metadata - this is what getVerbs() uses to find verbs
|
||
// Without this, getVerbsBySource() will still find "deleted" verbs via their metadata
|
||
try {
|
||
const metadata = await this.getVerbMetadata(id)
|
||
if (metadata) {
|
||
const verbType = (metadata.verb || metadata.type || 'default') as string
|
||
this.decrementVerbCount(verbType)
|
||
await this.deleteVerbMetadata(id)
|
||
}
|
||
} catch (error) {
|
||
// Ignore metadata deletion errors - verb file is already deleted
|
||
console.warn(`Failed to delete verb metadata for ${id}:`, 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
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 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
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 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: COW files (refs, blobs, commits - raw compressed)
|
||
// - .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')) {
|
||
// COW files 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
|
||
}
|
||
}
|
||
|
||
// ===========================================================================
|
||
// 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 cortex'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 entire branches/ directory (branch-based storage)
|
||
// Bug fix: Data is stored in branches/main/entities/, not just entities/
|
||
// The branch-based structure was introduced for COW support
|
||
const branchesDir = path.join(this.rootDir, 'branches')
|
||
if (await this.directoryExists(branchesDir)) {
|
||
await removeDirectoryContents(branchesDir)
|
||
}
|
||
|
||
// Remove all files in both system directories
|
||
await removeDirectoryContents(this.systemDir)
|
||
if (await this.directoryExists(this.indexDir)) {
|
||
await removeDirectoryContents(this.indexDir)
|
||
}
|
||
|
||
// Remove COW (copy-on-write) version control data
|
||
// This directory stores all git-like versioning data (commits, trees, blobs, refs)
|
||
// Must be deleted to fully clear all data including version history
|
||
const cowDir = path.join(this.rootDir, '_cow')
|
||
if (await this.directoryExists(cowDir)) {
|
||
// Delete the entire _cow/ directory (not just contents)
|
||
await fs.promises.rm(cowDir, { recursive: true, force: true })
|
||
|
||
// Reset COW managers (but don't disable COW - it's always enabled)
|
||
// COW will re-initialize automatically on next use
|
||
this.refManager = undefined
|
||
this.blobStorage = undefined
|
||
this.commitLog = undefined
|
||
}
|
||
|
||
// Clear the statistics cache
|
||
this.statisticsCache = null
|
||
this.statisticsModified = false
|
||
|
||
// Reset entity counters (inherited from BaseStorageAdapter)
|
||
// These in-memory counters must be reset to 0 after clearing all data
|
||
;(this as any).totalNounCount = 0
|
||
;(this as any).totalVerbCount = 0
|
||
|
||
// Clear write-through cache (inherited from BaseStorage)
|
||
// Without this, readWithInheritance() would return stale cached data
|
||
// after clear(), causing "ghost" entities to appear
|
||
this.clearWriteCache()
|
||
}
|
||
|
||
/**
|
||
* 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
|
||
}
|
||
|
||
/**
|
||
* Check if COW has been explicitly disabled via clear()
|
||
* Fixes bug where clear() doesn't persist across instance restarts
|
||
* @returns true if marker file exists, false otherwise
|
||
* @protected
|
||
*/
|
||
/**
|
||
* Removed checkClearMarker() and createClearMarker() methods
|
||
* COW is now always enabled - marker files are no longer used
|
||
*/
|
||
|
||
/**
|
||
* 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 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 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) {
|
||
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', rootDirectory: '${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 }.`
|
||
)
|
||
;(err as any).code = 'BRAINY_WRITER_LOCKED'
|
||
;(err as any).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 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 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 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 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 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 {
|
||
// CRITICAL: Detect existing depth before counting
|
||
// Can't use getAllShardedFiles() which assumes depth=1
|
||
const existingDepth = await this.detectExistingShardingDepth()
|
||
const depthToUse = existingDepth !== null ? existingDepth : this.SHARDING_DEPTH
|
||
|
||
// Count nouns using detected depth
|
||
const validNounFiles = await this.getAllFilesAtDepth(this.nounsDir, depthToUse)
|
||
this.totalNounCount = validNounFiles.length
|
||
|
||
// Count verbs using detected depth
|
||
const validVerbFiles = await this.getAllFilesAtDepth(this.verbsDir, depthToUse)
|
||
this.totalVerbCount = validVerbFiles.length
|
||
|
||
// Sample some files to get type distribution (don't read all)
|
||
// Load metadata separately for type information
|
||
const sampleSize = Math.min(100, validNounFiles.length)
|
||
for (let i = 0; i < sampleSize; i++) {
|
||
try {
|
||
const file = validNounFiles[i]
|
||
const id = file.replace('.json', '')
|
||
|
||
// Load metadata from separate storage for type info
|
||
const metadata = await this.getNounMetadata(id)
|
||
if (metadata) {
|
||
const type = metadata.noun || 'default'
|
||
this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1)
|
||
}
|
||
} catch {
|
||
// Skip invalid files or missing metadata
|
||
}
|
||
}
|
||
|
||
// Extrapolate counts if we sampled
|
||
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)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 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 files from one sharding depth to another
|
||
* Handles: 0→1 (flat to single-level), 2→1 (deep to single-level)
|
||
* Uses atomic file operations and comprehensive error handling
|
||
*
|
||
* @param fromDepth - Source sharding depth
|
||
* @param toDepth - Target sharding depth (must be 1)
|
||
*/
|
||
private async migrateShardingStructure(fromDepth: number, toDepth: number): Promise<void> {
|
||
// Validation
|
||
if (fromDepth === toDepth) {
|
||
throw new Error(`Migration not needed: already at depth ${toDepth}`)
|
||
}
|
||
|
||
if (toDepth !== 1) {
|
||
throw new Error(`Migration only supports target depth 1 (got ${toDepth})`)
|
||
}
|
||
|
||
if (fromDepth !== 0 && fromDepth !== 2) {
|
||
throw new Error(`Migration only supports source depth 0 or 2 (got ${fromDepth})`)
|
||
}
|
||
|
||
// Create migration lock to prevent concurrent migrations
|
||
const lockFile = path.join(this.systemDir, '.migration-lock')
|
||
const lockExists = await this.fileExists(lockFile)
|
||
|
||
if (lockExists) {
|
||
// Check if lock is stale (> 1 hour old)
|
||
try {
|
||
const stats = await fs.promises.stat(lockFile)
|
||
const lockAge = Date.now() - stats.mtimeMs
|
||
const ONE_HOUR = 60 * 60 * 1000
|
||
|
||
if (lockAge < ONE_HOUR) {
|
||
throw new Error(
|
||
'Migration already in progress. If this is incorrect, delete .migration-lock file.'
|
||
)
|
||
}
|
||
|
||
// Lock is stale, remove it
|
||
console.log('⚠️ Removing stale migration lock (> 1 hour old)')
|
||
await fs.promises.unlink(lockFile)
|
||
} catch (error: any) {
|
||
if (error.code !== 'ENOENT') {
|
||
throw error
|
||
}
|
||
}
|
||
}
|
||
|
||
try {
|
||
// Create lock file
|
||
await fs.promises.writeFile(lockFile, JSON.stringify({
|
||
startedAt: new Date().toISOString(),
|
||
fromDepth,
|
||
toDepth,
|
||
pid: process.pid
|
||
}))
|
||
|
||
// Discover all files to migrate
|
||
console.log('📊 Discovering files to migrate...')
|
||
const filesToMigrate = await this.discoverFilesForMigration(fromDepth)
|
||
|
||
if (filesToMigrate.length === 0) {
|
||
console.log('ℹ️ No files to migrate')
|
||
return
|
||
}
|
||
|
||
console.log(`📦 Migrating ${filesToMigrate.length} files...`)
|
||
|
||
// Create all target shard directories upfront
|
||
await this.createAllShardDirectories(this.nounsDir)
|
||
await this.createAllShardDirectories(this.verbsDir)
|
||
|
||
// Migrate files with progress tracking
|
||
let migratedCount = 0
|
||
let skippedCount = 0
|
||
const errors: Array<{ file: string; error: string }> = []
|
||
|
||
for (const fileInfo of filesToMigrate) {
|
||
try {
|
||
await this.migrateFile(fileInfo, fromDepth, toDepth)
|
||
migratedCount++
|
||
|
||
// Progress update every 1000 files
|
||
if (migratedCount % 1000 === 0) {
|
||
const percent = ((migratedCount / filesToMigrate.length) * 100).toFixed(1)
|
||
console.log(` 📊 Progress: ${migratedCount}/${filesToMigrate.length} (${percent}%)`)
|
||
}
|
||
|
||
// Yield to event loop every 100 files to prevent blocking
|
||
if (migratedCount % 100 === 0) {
|
||
await new Promise(resolve => setImmediate(resolve))
|
||
}
|
||
} catch (error: any) {
|
||
skippedCount++
|
||
errors.push({
|
||
file: fileInfo.oldPath,
|
||
error: error.message
|
||
})
|
||
|
||
// Log first few errors
|
||
if (errors.length <= 5) {
|
||
console.warn(`⚠️ Skipped ${fileInfo.oldPath}: ${error.message}`)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Final summary
|
||
console.log(`\n✅ Migration Results:`)
|
||
console.log(` Migrated: ${migratedCount} files`)
|
||
console.log(` Skipped: ${skippedCount} files`)
|
||
|
||
if (errors.length > 0) {
|
||
console.warn(`\n⚠️ ${errors.length} files could not be migrated`)
|
||
if (errors.length > 5) {
|
||
console.warn(` (First 5 errors shown above, ${errors.length - 5} more occurred)`)
|
||
}
|
||
}
|
||
|
||
// Cleanup: Remove empty old directories
|
||
if (fromDepth === 0) {
|
||
// No subdirectories to clean for flat structure
|
||
} else if (fromDepth === 2) {
|
||
await this.cleanupEmptyDirectories(this.nounsDir, fromDepth)
|
||
await this.cleanupEmptyDirectories(this.verbsDir, fromDepth)
|
||
}
|
||
|
||
// Verification: Count files in new structure
|
||
const verifyCount = await this.countFilesInStructure(toDepth)
|
||
console.log(`\n🔍 Verification: ${verifyCount} files in new structure`)
|
||
|
||
if (verifyCount < migratedCount) {
|
||
console.warn(`⚠️ Warning: Verification count (${verifyCount}) < migrated count (${migratedCount})`)
|
||
}
|
||
} finally {
|
||
// Always remove lock file
|
||
try {
|
||
await fs.promises.unlink(lockFile)
|
||
} catch (error) {
|
||
// Ignore error if lock file doesn't exist
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Discover all files that need to be migrated
|
||
* Constructs correct oldPath based on source depth
|
||
*/
|
||
private async discoverFilesForMigration(fromDepth: number): Promise<Array<{ oldPath: string; id: string; type: 'noun' | 'verb' }>> {
|
||
const files: Array<{ oldPath: string; id: string; type: 'noun' | 'verb' }> = []
|
||
|
||
// Discover noun files
|
||
const nounFiles = await this.getAllFilesAtDepth(this.nounsDir, fromDepth)
|
||
for (const filename of nounFiles) {
|
||
const id = filename.replace('.json', '')
|
||
|
||
// Construct correct oldPath based on fromDepth
|
||
let oldPath: string
|
||
switch (fromDepth) {
|
||
case 0:
|
||
// Flat: nouns/uuid.json
|
||
oldPath = path.join(this.nounsDir, `${id}.json`)
|
||
break
|
||
case 1:
|
||
// Single-level: nouns/ab/uuid.json
|
||
oldPath = path.join(this.nounsDir, id.substring(0, 2), `${id}.json`)
|
||
break
|
||
case 2:
|
||
// Deep: nouns/ab/cd/uuid.json
|
||
oldPath = path.join(this.nounsDir, id.substring(0, 2), id.substring(2, 4), `${id}.json`)
|
||
break
|
||
default:
|
||
throw new Error(`Unsupported fromDepth: ${fromDepth}`)
|
||
}
|
||
|
||
files.push({ oldPath, id, type: 'noun' })
|
||
}
|
||
|
||
// Discover verb files
|
||
const verbFiles = await this.getAllFilesAtDepth(this.verbsDir, fromDepth)
|
||
for (const filename of verbFiles) {
|
||
const id = filename.replace('.json', '')
|
||
|
||
// Construct correct oldPath based on fromDepth
|
||
let oldPath: string
|
||
switch (fromDepth) {
|
||
case 0:
|
||
// Flat: verbs/uuid.json
|
||
oldPath = path.join(this.verbsDir, `${id}.json`)
|
||
break
|
||
case 1:
|
||
// Single-level: verbs/ab/uuid.json
|
||
oldPath = path.join(this.verbsDir, id.substring(0, 2), `${id}.json`)
|
||
break
|
||
case 2:
|
||
// Deep: verbs/ab/cd/uuid.json
|
||
oldPath = path.join(this.verbsDir, id.substring(0, 2), id.substring(2, 4), `${id}.json`)
|
||
break
|
||
default:
|
||
throw new Error(`Unsupported fromDepth: ${fromDepth}`)
|
||
}
|
||
|
||
files.push({ oldPath, id, type: 'verb' })
|
||
}
|
||
|
||
return files
|
||
}
|
||
|
||
/**
|
||
* Get all files at a specific depth
|
||
*/
|
||
private async getAllFilesAtDepth(baseDir: string, depth: number): Promise<string[]> {
|
||
const allFiles: string[] = []
|
||
|
||
try {
|
||
const dirExists = await this.directoryExists(baseDir)
|
||
if (!dirExists) {
|
||
return []
|
||
}
|
||
|
||
switch (depth) {
|
||
case 0:
|
||
// Flat: files directly in baseDir
|
||
const entries = await fs.promises.readdir(baseDir)
|
||
for (const entry of entries) {
|
||
if (entry.endsWith('.json')) {
|
||
allFiles.push(entry)
|
||
}
|
||
}
|
||
break
|
||
|
||
case 1:
|
||
// Single-level: baseDir/ab/uuid.json
|
||
const shardDirs = await fs.promises.readdir(baseDir)
|
||
for (const shard of shardDirs) {
|
||
const shardPath = path.join(baseDir, shard)
|
||
try {
|
||
const stat = await fs.promises.stat(shardPath)
|
||
if (stat.isDirectory()) {
|
||
const shardFiles = await fs.promises.readdir(shardPath)
|
||
for (const file of shardFiles) {
|
||
if (file.endsWith('.json')) {
|
||
allFiles.push(file)
|
||
}
|
||
}
|
||
}
|
||
} catch (error) {
|
||
// Skip inaccessible directories
|
||
}
|
||
}
|
||
break
|
||
|
||
case 2:
|
||
// Deep: baseDir/ab/cd/uuid.json
|
||
const level1Dirs = await fs.promises.readdir(baseDir)
|
||
for (const level1 of level1Dirs) {
|
||
const level1Path = path.join(baseDir, level1)
|
||
try {
|
||
const level1Stat = await fs.promises.stat(level1Path)
|
||
if (level1Stat.isDirectory()) {
|
||
const level2Dirs = await fs.promises.readdir(level1Path)
|
||
for (const level2 of level2Dirs) {
|
||
const level2Path = path.join(level1Path, level2)
|
||
try {
|
||
const level2Stat = await fs.promises.stat(level2Path)
|
||
if (level2Stat.isDirectory()) {
|
||
const files = await fs.promises.readdir(level2Path)
|
||
for (const file of files) {
|
||
if (file.endsWith('.json')) {
|
||
allFiles.push(file)
|
||
}
|
||
}
|
||
}
|
||
} catch (error) {
|
||
// Skip inaccessible directories
|
||
}
|
||
}
|
||
}
|
||
} catch (error) {
|
||
// Skip inaccessible directories
|
||
}
|
||
}
|
||
break
|
||
}
|
||
} catch (error) {
|
||
// Directory doesn't exist or not accessible
|
||
}
|
||
|
||
return allFiles
|
||
}
|
||
|
||
/**
|
||
* Create all 256 shard directories (00-ff)
|
||
*/
|
||
private async createAllShardDirectories(baseDir: string): Promise<void> {
|
||
for (let i = 0; i < this.MAX_SHARDS; i++) {
|
||
const shard = i.toString(16).padStart(2, '0')
|
||
const shardDir = path.join(baseDir, shard)
|
||
await this.ensureDirectoryExists(shardDir)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 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)
|
||
}
|
||
|
||
/**
|
||
* Clean up empty directories after migration
|
||
*/
|
||
private async cleanupEmptyDirectories(baseDir: string, depth: number): Promise<void> {
|
||
try {
|
||
if (depth === 2) {
|
||
// Clean up level2 and level1 directories
|
||
const level1Dirs = await fs.promises.readdir(baseDir)
|
||
for (const level1 of level1Dirs) {
|
||
const level1Path = path.join(baseDir, level1)
|
||
try {
|
||
const level1Stat = await fs.promises.stat(level1Path)
|
||
if (level1Stat.isDirectory()) {
|
||
const level2Dirs = await fs.promises.readdir(level1Path)
|
||
for (const level2 of level2Dirs) {
|
||
const level2Path = path.join(level1Path, level2)
|
||
try {
|
||
// Try to remove level2 directory (will fail if not empty)
|
||
await fs.promises.rmdir(level2Path)
|
||
} catch (error) {
|
||
// Directory not empty or other error - ignore
|
||
}
|
||
}
|
||
|
||
// Try to remove level1 directory
|
||
await fs.promises.rmdir(level1Path)
|
||
}
|
||
} catch (error) {
|
||
// Directory not empty or other error - ignore
|
||
}
|
||
}
|
||
}
|
||
} catch (error) {
|
||
// Cleanup is best-effort, don't throw
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Count files in the current structure
|
||
*/
|
||
private async countFilesInStructure(depth: number): Promise<number> {
|
||
let count = 0
|
||
|
||
count += (await this.getAllFilesAtDepth(this.nounsDir, depth)).length
|
||
count += (await this.getAllFilesAtDepth(this.verbsDir, depth)).length
|
||
|
||
return count
|
||
}
|
||
|
||
/**
|
||
* Detect the actual sharding depth used by existing files
|
||
* Examines directory structure to determine current sharding strategy
|
||
* Returns null if no files exist yet (new installation)
|
||
*/
|
||
private async detectExistingShardingDepth(): Promise<number | null> {
|
||
try {
|
||
// Check if nouns directory exists and has content
|
||
const dirExists = await this.directoryExists(this.nounsDir)
|
||
if (!dirExists) {
|
||
return null // New installation
|
||
}
|
||
|
||
const entries = await fs.promises.readdir(this.nounsDir, { withFileTypes: true })
|
||
|
||
// Check if there are any .json files directly in nounsDir (flat structure)
|
||
const hasDirectJsonFiles = entries.some((e: any) => e.isFile() && e.name.endsWith('.json'))
|
||
if (hasDirectJsonFiles) {
|
||
return 0 // Flat structure: nouns/uuid.json
|
||
}
|
||
|
||
// Check for subdirectories with hex names (sharding directories)
|
||
const subdirs = entries.filter((e: any) => e.isDirectory() && /^[0-9a-f]{2}$/i.test(e.name))
|
||
if (subdirs.length === 0) {
|
||
return null // No files yet
|
||
}
|
||
|
||
// Check first subdir to see if it has files or more subdirs
|
||
const firstSubdir = subdirs[0].name
|
||
const subdirPath = path.join(this.nounsDir, firstSubdir)
|
||
const subdirEntries = await fs.promises.readdir(subdirPath, { withFileTypes: true })
|
||
|
||
const hasJsonFiles = subdirEntries.some((e: any) => e.isFile() && e.name.endsWith('.json'))
|
||
if (hasJsonFiles) {
|
||
return 1 // Single-level sharding: nouns/ab/uuid.json
|
||
}
|
||
|
||
const hasSubSubdirs = subdirEntries.some((e: any) => e.isDirectory() && /^[0-9a-f]{2}$/i.test(e.name))
|
||
if (hasSubSubdirs) {
|
||
return 2 // Deep sharding: nouns/ab/cd/uuid.json
|
||
}
|
||
|
||
return 1 // Default to single-level if structure is unclear
|
||
} catch (error) {
|
||
// If we can't read the directory, assume new installation
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get sharding depth
|
||
* Always returns 1 (single-level sharding) for optimal balance of
|
||
* simplicity, performance, and reliability across all dataset sizes
|
||
*
|
||
* Single-level sharding (depth=1):
|
||
* - 256 shard directories (00-ff)
|
||
* - Handles 2.5M+ entities with excellent performance
|
||
* - No dynamic depth changes = no path mismatch bugs
|
||
* - Industry standard approach (Git uses similar)
|
||
*/
|
||
private getOptimalShardingDepth(): number {
|
||
return this.SHARDING_DEPTH
|
||
}
|
||
|
||
/**
|
||
* Get the path for a node with consistent sharding strategy
|
||
* Clean, predictable path generation
|
||
*/
|
||
private getNodePath(id: string): string {
|
||
return this.getShardedPath(this.nounsDir, id)
|
||
}
|
||
|
||
/**
|
||
* Get the path for a verb with consistent sharding strategy
|
||
*/
|
||
private getVerbPath(id: string): string {
|
||
return this.getShardedPath(this.verbsDir, id)
|
||
}
|
||
|
||
/**
|
||
* Universal sharded path generator
|
||
* Always uses depth=1 (single-level sharding) for consistency
|
||
*
|
||
* Format: baseDir/ab/uuid.json
|
||
* Where 'ab' = first 2 hex characters of UUID (lowercase)
|
||
*
|
||
* Validates UUID format and throws descriptive errors
|
||
*/
|
||
private getShardedPath(baseDir: string, id: string): string {
|
||
// Extract first 2 characters for shard directory
|
||
const shard = id.substring(0, 2).toLowerCase()
|
||
|
||
// Validate shard is valid hex (00-ff)
|
||
if (!/^[0-9a-f]{2}$/.test(shard)) {
|
||
throw new Error(
|
||
`Invalid entity ID format: ${id}. ` +
|
||
`Expected UUID starting with 2 hex characters, got '${shard}'. ` +
|
||
`IDs must be UUIDs or hex strings.`
|
||
)
|
||
}
|
||
|
||
// Single-level sharding: baseDir/ab/uuid.json
|
||
return path.join(baseDir, shard, `${id}.json`)
|
||
}
|
||
|
||
/**
|
||
* Get all JSON files from the single-level sharded directory structure
|
||
* Traverses all shard subdirectories (00-ff)
|
||
*/
|
||
private async getAllShardedFiles(baseDir: string): Promise<string[]> {
|
||
const allFiles: string[] = []
|
||
|
||
try {
|
||
const shardDirs = await fs.promises.readdir(baseDir)
|
||
|
||
for (const shardDir of shardDirs) {
|
||
const shardPath = path.join(baseDir, shardDir)
|
||
|
||
try {
|
||
const stat = await fs.promises.stat(shardPath)
|
||
|
||
if (stat.isDirectory()) {
|
||
const shardFiles = await fs.promises.readdir(shardPath)
|
||
for (const file of shardFiles) {
|
||
if (file.endsWith('.json')) {
|
||
allFiles.push(file)
|
||
}
|
||
}
|
||
}
|
||
} catch (shardError) {
|
||
// Skip inaccessible shard directories
|
||
continue
|
||
}
|
||
}
|
||
|
||
// Sort for consistent ordering
|
||
allFiles.sort()
|
||
return allFiles
|
||
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') {
|
||
// Directory doesn't exist yet
|
||
return []
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Production-scale streaming pagination for very large datasets
|
||
* Avoids loading all filenames into memory
|
||
*/
|
||
private async getVerbsWithPaginationStreaming(
|
||
options: {
|
||
limit?: number
|
||
cursor?: string
|
||
filter?: {
|
||
verbType?: string | string[]
|
||
sourceId?: string | string[]
|
||
targetId?: string | string[]
|
||
service?: string | string[]
|
||
metadata?: Record<string, any>
|
||
}
|
||
},
|
||
startIndex: number,
|
||
limit: number
|
||
): Promise<{
|
||
items: HNSWVerbWithMetadata[]
|
||
totalCount?: number
|
||
hasMore: boolean
|
||
nextCursor?: string
|
||
}> {
|
||
const verbs: HNSWVerbWithMetadata[] = []
|
||
let processedCount = 0
|
||
let skippedCount = 0
|
||
let resultCount = 0
|
||
|
||
const depth = this.cachedShardingDepth ?? this.getOptimalShardingDepth()
|
||
|
||
try {
|
||
// Stream through sharded directories efficiently
|
||
// hasMore=false means we reached the end of files, hasMore=true means streaming stopped early
|
||
const streamingHasMore = await this.streamShardedFiles(
|
||
this.verbsDir,
|
||
depth,
|
||
async (filename: string, filePath: string) => {
|
||
// Skip files until we reach start index
|
||
if (skippedCount < startIndex) {
|
||
skippedCount++
|
||
return true // continue
|
||
}
|
||
|
||
// Stop if we have enough results
|
||
if (resultCount >= limit) {
|
||
return false // stop streaming - more files exist
|
||
}
|
||
|
||
try {
|
||
const id = filename.replace('.json', '')
|
||
|
||
// Read verb data and metadata
|
||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||
const edge = JSON.parse(data)
|
||
const metadata = await this.getVerbMetadata(id)
|
||
|
||
// Don't skip verbs without metadata - metadata is optional
|
||
// FIX: This was the root cause of the VFS bug (11 versions)
|
||
// Verbs can exist without metadata files (e.g., from imports/migrations)
|
||
|
||
// Convert connections if needed
|
||
let connections = edge.connections
|
||
if (connections && typeof connections === 'object' && !(connections instanceof Map)) {
|
||
const connectionsMap = new Map<number, Set<string>>()
|
||
for (const [level, nodeIds] of Object.entries(connections)) {
|
||
connectionsMap.set(Number(level), new Set(nodeIds as string[]))
|
||
}
|
||
connections = connectionsMap
|
||
}
|
||
|
||
// Extract standard fields from metadata to top-level
|
||
const metadataObj = (metadata || {}) as VerbMetadata
|
||
const { createdAt, updatedAt, confidence, weight, service, data: dataField, createdBy, ...customMetadata } = metadataObj
|
||
|
||
const verbWithMetadata: HNSWVerbWithMetadata = {
|
||
id: edge.id,
|
||
vector: edge.vector,
|
||
connections: connections || new Map(),
|
||
verb: edge.verb,
|
||
sourceId: edge.sourceId,
|
||
targetId: edge.targetId,
|
||
createdAt: (createdAt as number) || Date.now(),
|
||
updatedAt: (updatedAt as number) || Date.now(),
|
||
confidence: confidence as number | undefined,
|
||
weight: weight as number | undefined,
|
||
service: service as string | undefined,
|
||
data: dataField as Record<string, any> | undefined,
|
||
createdBy,
|
||
metadata: customMetadata
|
||
}
|
||
|
||
// Apply filters
|
||
if (options.filter) {
|
||
const filter = options.filter
|
||
|
||
if (filter.verbType) {
|
||
const types = Array.isArray(filter.verbType) ? filter.verbType : [filter.verbType]
|
||
if (!types.includes(verbWithMetadata.verb)) return true // continue
|
||
}
|
||
|
||
if (filter.sourceId) {
|
||
const sources = Array.isArray(filter.sourceId) ? filter.sourceId : [filter.sourceId]
|
||
if (!sources.includes(verbWithMetadata.sourceId)) return true // continue
|
||
}
|
||
|
||
if (filter.targetId) {
|
||
const targets = Array.isArray(filter.targetId) ? filter.targetId : [filter.targetId]
|
||
if (!targets.includes(verbWithMetadata.targetId)) return true // continue
|
||
}
|
||
}
|
||
|
||
verbs.push(verbWithMetadata)
|
||
resultCount++
|
||
processedCount++
|
||
return true // continue
|
||
|
||
} catch (error) {
|
||
console.warn(`Failed to read verb from ${filePath}:`, error)
|
||
processedCount++
|
||
return true // continue
|
||
}
|
||
}
|
||
)
|
||
|
||
// CRITICAL FIX: Use streaming result for hasMore, not cached totalVerbCount
|
||
// streamingHasMore=false means we exhausted all files
|
||
// Also verify we loaded items to prevent infinite loops
|
||
const finalHasMore = streamingHasMore && (resultCount > 0 || startIndex === 0)
|
||
|
||
return {
|
||
items: verbs,
|
||
totalCount: this.totalVerbCount || undefined, // Return cached count as hint only
|
||
hasMore: finalHasMore,
|
||
nextCursor: finalHasMore ? String(startIndex + resultCount) : undefined
|
||
}
|
||
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') {
|
||
return {
|
||
items: [],
|
||
totalCount: 0,
|
||
hasMore: false
|
||
}
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Stream through sharded files without loading all names into memory
|
||
* Production-scale implementation for millions of files
|
||
*/
|
||
/**
|
||
* Stream through files in single-level sharded structure
|
||
* Calls processor for each file until processor returns false
|
||
* Returns true if more files exist (processor stopped early), false if all processed
|
||
*/
|
||
private async streamShardedFiles(
|
||
baseDir: string,
|
||
depth: number,
|
||
processor: (filename: string, fullPath: string) => Promise<boolean>
|
||
): Promise<boolean> {
|
||
let hasMore = true
|
||
|
||
// Single-level sharding (depth=1): baseDir/ab/uuid.json
|
||
try {
|
||
const shardDirs = await fs.promises.readdir(baseDir)
|
||
const sortedShardDirs = shardDirs.sort()
|
||
|
||
for (const shardDir of sortedShardDirs) {
|
||
const shardPath = path.join(baseDir, shardDir)
|
||
|
||
try {
|
||
const stat = await fs.promises.stat(shardPath)
|
||
|
||
if (stat.isDirectory()) {
|
||
const files = await fs.promises.readdir(shardPath)
|
||
const sortedFiles = files.filter((f: string) => f.endsWith('.json')).sort()
|
||
|
||
for (const file of sortedFiles) {
|
||
const shouldContinue = await processor(file, path.join(shardPath, file))
|
||
|
||
if (!shouldContinue) {
|
||
hasMore = false
|
||
break
|
||
}
|
||
}
|
||
|
||
if (!hasMore) break
|
||
}
|
||
} catch (shardError) {
|
||
// Skip inaccessible shard directories
|
||
continue
|
||
}
|
||
}
|
||
} catch (error: any) {
|
||
if (error.code === 'ENOENT') {
|
||
hasMore = false
|
||
}
|
||
}
|
||
|
||
return hasMore
|
||
}
|
||
|
||
/**
|
||
* 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 saveHNSWData(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 getHNSWData(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
|
||
}
|
||
}
|
||
}
|