mpackdb

35.5 KB

1. import { dirname, basename, extname, resolve } from 'node:path';
2. import { createReadStream, createWriteStream } from 'node:fs';
3. import { mkdir, open, stat, readFile, writeFile, rename, unlink } from 'node:fs/promises';
4. import { AsyncLocalStorage } from 'node:async_hooks';
5. import { serialize, deserialize, uuid, PrimaryKeyType, IndexType } from './mpack.js';
6. import { Cursor } from './Cursor.js';
7. import { IndexManager } from './IndexManager.js';
9. // Tracks which async call chain currently owns a db's lock, so nested operations
10. // (e.g. find inside delete, insert inside withLock) re-enter, while unrelated
11. // concurrent operations queue up instead of walking into the critical section.
12. const lockContext = new AsyncLocalStorage();
14. export { PrimaryKeyType, IndexType };
16. /**
17. * MPackDB - A fast, local, append-only JSON database with MessagePack serialization
18. *
19. * Features:
20. * - Append-only writes for high performance
21. * - MessagePack binary serialization
22. * - Optional indexes (numeric and lexical)
23. * - File-based locking for concurrent access
24. * - Auto-compaction on startup
25. * - Auto-persistence of indexes
26. */
27. export class MPackDB {
28. _dbFile = null;
29. _primaryKeyType = null;
30. _primaryKey = null;
31. _classToUse = null;
33. _indexes = [];
34. _indexTypes = {};
35. _uniqueIndexes = new Set();
37. _initPromise = null;
38. _initState = 0;
39. _dataPath = null;
40. _dataStream = null;
41. _dataIno = null;
42. _lastWrite = null;
43. _lockPath = null;
44. _mutexTail = Promise.resolve();
45. _staleLockTimeout = 30000;
46. _schema = null;
47. _indexManager = null;
48. _meta = {
49. nextId: 0,
50. deleted: [],
51. };
52. _debug = false;
53. _indexPersistInterval = 60000; // 60 seconds default
54. _indexPersistThreshold = 1000; // 1000 entries default
55. _processExitHandler = null;
57. /**
58. * Create a new MPackDB instance
59. *
60. * @param {string} dbFile - Path to the database file (without extension)
61. * @param {Object} options - Configuration options
62. * @param {string} [options.primaryKey] - Primary key field name. Prefix with * for numeric (e.g., '*id'), @ for UUID (e.g., '@uuid'), or no prefix for string
63. * @param {PrimaryKeyType} [options.primaryKeyType] - Explicit primary key type (overrides prefix)
64. * @param {string[]} [options.indexes] - Array of field names to index. Use * prefix for numeric, @ for UUID
65. * @param {boolean} [options.debug=false] - Enable debug logging
66. * @param {number} [options.indexPersistInterval=60000] - Milliseconds between automatic index persistence (0 to disable)
67. * @param {number} [options.indexPersistThreshold=1000] - Number of changes before auto-persisting indexes
68. * @param {boolean} [options.compact=true] - Run compaction on init (set false whenever another process may have the same files open)
69. * @param {number} [options.staleLockTimeout=30000] - Take over lock files older than this many ms (crashed holder); 0 disables
70. *
71. * @example
72. * const db = new MPackDB('data/users', {
73. * primaryKey: '*id', // Numeric auto-increment
74. * indexes: ['email', '*age'], // Index email (lexical) and age (numeric)
75. * indexPersistThreshold: 100
76. * });
77. */
78. _compact = true;
80. _hasPrimaryKeyValue(record) {
81. if (!this._primaryKey || !record || !Object.prototype.hasOwnProperty.call(record, this._primaryKey)) {
82. return false;
83. }
84. const value = record[this._primaryKey];
85. return value !== undefined && value !== null;
86. }
88. constructor(dbFile, { primaryKey, primaryKeyType, indexes, debug, indexPersistInterval, indexPersistThreshold, compact, staleLockTimeout } = {}) {
89. this._dbFile = dbFile || this._dbFile;
90. this._debug = debug || this._debug;
91. if (compact !== undefined) this._compact = compact;
92. if (indexPersistInterval !== undefined) this._indexPersistInterval = indexPersistInterval;
93. if (indexPersistThreshold !== undefined) this._indexPersistThreshold = indexPersistThreshold;
94. if (staleLockTimeout !== undefined) this._staleLockTimeout = staleLockTimeout;
96. // Parse primary key with optional prefix
97. if (primaryKey) {
98. if (primaryKey.startsWith('*')) {
99. // *id = numeric primary key
100. this._primaryKey = primaryKey.slice(1);
101. this._primaryKeyType = PrimaryKeyType.NUMBER;
102. } else if (primaryKey.startsWith('@')) {
103. // @id = UUID primary key
104. this._primaryKey = primaryKey.slice(1);
105. this._primaryKeyType = PrimaryKeyType.UUID;
106. } else {
107. // id = string primary key (lexical)
108. this._primaryKey = primaryKey;
109. this._primaryKeyType = PrimaryKeyType.STRING;
110. }
111. // Allow explicit override
112. if (primaryKeyType !== undefined) {
113. this._primaryKeyType = primaryKeyType;
114. }
115. }
117. // Primary key is always unique
118. if (this._primaryKey) {
119. this._uniqueIndexes.add(this._primaryKey);
120. }
122. const allIndexes = [...new Set([
123. ...(this._primaryKey ? [this._primaryKey] : []),
124. ...(indexes || []),
125. ...(this._indexes || []),
126. ])];
127. this._indexes = [];
128. for (let index of allIndexes) {
129. let cleanIndex = index;
130. // !field = unique index (can combine with * and @: !*field, !@field)
131. let isUnique = false;
132. if (cleanIndex.startsWith('!')) {
133. isUnique = true;
134. cleanIndex = cleanIndex.slice(1);
135. }
136. if (cleanIndex.startsWith('*')) {
137. // *field = numeric index
138. cleanIndex = cleanIndex.slice(1);
139. this._indexTypes[cleanIndex] = IndexType.NUMERIC;
140. } else if (cleanIndex.startsWith('@')) {
141. // @field = UUID index (lexical)
142. cleanIndex = cleanIndex.slice(1);
143. this._indexTypes[cleanIndex] = IndexType.LEXICAL;
144. } else if (cleanIndex === this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER) {
145. // Primary key is numeric
146. this._indexTypes[cleanIndex] = IndexType.NUMERIC;
147. } else {
148. // Default to lexical
149. this._indexTypes[cleanIndex] = IndexType.LEXICAL;
150. }
151. if (isUnique) {
152. this._uniqueIndexes.add(cleanIndex);
153. }
154. this._indexes.push(cleanIndex);
155. }
157. // Schema descriptor persisted into meta.json so other tools (e.g. mpackdb-admin)
158. // can open this database with the correct primary key and indexes.
159. if (this._primaryKey || this._indexes.length > 0) {
160. const pkPrefix = this._primaryKeyType === PrimaryKeyType.NUMBER ? '*'
161. : this._primaryKeyType === PrimaryKeyType.UUID ? '@' : '';
162. this._schema = {
163. primaryKey: this._primaryKey ? pkPrefix + this._primaryKey : undefined,
164. indexes: this._indexes
165. .filter(field => field !== this._primaryKey)
166. .map(field =>
167. (this._uniqueIndexes.has(field) ? '!' : '')
168. + (this._indexTypes[field] === IndexType.NUMERIC ? '*' : '')
169. + field
170. ),
171. };
172. }
173. }
175. /**
176. * Initialize the database (called automatically by other methods)
177. * Performs compaction and loads metadata
178. *
179. * @returns {Promise<MPackDB>} The database instance
180. */
181. async init() {
182. if (this._initState === 2) return this;
184. if (this._initState === 1) {
185. return this._initPromise;
186. }
188. this._initState = 1;
189. return this._initPromise = new Promise(async success => {
190. if (!this._dbFile) {
191. throw new Error('No database file specified.');
192. }
194. const dbDir = dirname(this._dbFile);
195. const baseName = basename(this._dbFile, extname(this._dbFile));
196. await mkdir(dbDir, { recursive: true });
198. this._dataPath = resolve(dbDir, `${baseName}.mpack`);
199. this._metaPath = resolve(dbDir, `${baseName}.meta.json`);
200. this._lockPath = resolve(dbDir, `${baseName}.lock`);
202. try {
203. this._meta = JSON.parse(await readFile(this._metaPath));
204. } catch (e) {
205. this._meta = {
206. nextId: 0,
207. deleted: [],
208. };
209. }
211. const didCompact = this._compact;
212. if (didCompact) await this.compact({ duringInit: true });
214. this._dataStream = createWriteStream(this._dataPath, { flags: 'a' }); // needs to be set after compact as it replaces the file with a tmp file
215. this._dataIno = (await stat(this._dataPath).catch(() => null))?.ino ?? null;
217. // Persist schema into meta.json so other tools (mpackdb-admin) can
218. // open this db with the right primary key / indexes. The compare
219. // runs under the lock: refresh() has adopted any newer on-disk meta
220. // by then, so a concurrent process' meta is never clobbered.
221. if (this._schema) {
222. await this._withFileLock('init-schema', async () => {
223. if (JSON.stringify(this._meta.schema) !== JSON.stringify(this._schema)) {
224. await this.persistMeta();
225. }
226. });
227. }
229. // Initialize IndexManager if indexes are specified
230. if (this._indexes.length > 0) {
231. const dbDir = dirname(this._dbFile);
232. const baseName = basename(this._dbFile, extname(this._dbFile));
233. this._indexManager = new IndexManager(dbDir, baseName, this._indexes, this._indexTypes, this._primaryKeyType, {
234. // Index persistence rewrites shared files — must hold the db lock.
235. // lockContext.exit: a threshold-triggered persist inside insert()
236. // must NOT inherit insert's lock ownership (it outlives it), so it
237. // queues for its own turn instead.
238. runExclusive: fn => lockContext.exit(() => this._withFileLock('index-persist', fn)),
239. });
240. // Index files are shared with other processes — (re)build them under the lock
241. await this._withFileLock('index-init', () =>
242. this._indexManager.init(this._dataPath, { forceRebuild: didCompact })
243. );
244. // Start auto-persist with configured interval and threshold
245. this._indexManager.startAutoPersist(this._indexPersistInterval, this._indexPersistThreshold);
246. // Register signal handlers for Ctrl+C and kill signals
247. this._processExitHandler = async () => {
248. await this.close();
249. process.exit(0);
250. };
251. process.once('SIGINT', this._processExitHandler);
252. process.once('SIGTERM', this._processExitHandler);
253. }
255. this._initState = 2;
256. this._initPromise = null;
257. success(this);
258. });
259. }
261. /**
262. * Insert a new record into the database
263. *
264. * @param {Object} record - The record to insert
265. * @param {Object} [options]
266. * @param {boolean} [options.skipPrimaryKey=false] - Skip auto-generating primary key
267. * @returns {Promise<string|number|Object>} The primary-key value when configured, otherwise the inserted record
268. *
269. * @example
270. * const id = await db.insert({ name: 'Alice', age: 30 });
271. * // Returns: 0 (the generated primary-key value)
272. */
273. async insert(record, { skipPrimaryKey = false } = {}) {
274. await this.init();
275. return this._withFileLock('insert', async () => {
276. const recToInsert = this._classToUse
277. ? Object.assign(new this._classToUse(), record)
278. : { ...record };
280. const hasPrimaryKeyValue = this._hasPrimaryKeyValue(recToInsert);
281. if (this._primaryKey && !skipPrimaryKey && !hasPrimaryKeyValue) {
282. if (this._primaryKeyType === PrimaryKeyType.NUMBER) {
283. recToInsert[this._primaryKey] = this._meta.nextId++;
284. } else if (this._primaryKeyType === PrimaryKeyType.UUID) {
285. recToInsert[this._primaryKey] = uuid();
286. }
287. }
289. // Check unique index constraints (skip auto-generated primary keys — guaranteed unique)
290. if (this._indexManager && this._uniqueIndexes.size > 0) {
291. const autoGenPK = this._primaryKey && !skipPrimaryKey && !hasPrimaryKeyValue
292. && (this._primaryKeyType === PrimaryKeyType.NUMBER || this._primaryKeyType === PrimaryKeyType.UUID);
293. const deletedSet = new Set(this._meta.deleted);
294. for (const field of this._uniqueIndexes) {
295. if (autoGenPK && field === this._primaryKey) continue;
296. const value = recToInsert[field];
297. if (value === undefined) continue;
298. const existing = await this._indexManager.get(field, value);
299. const live = existing.filter(e => !deletedSet.has(e.loc[0]));
300. if (live.length > 0) {
301. const err = new Error(`Duplicate key: ${field}=${value}`);
302. err.code = 'DUPLICATE_KEY';
303. err.field = field;
304. err.value = value;
305. throw err;
306. }
307. }
308. }
310. const packedBuffer = serialize(recToInsert);
311. const fileStat = await stat(this._dataPath).catch(() => ({ size: 0 }));
312. const offset = fileStat.size;
314. const writePromise = new Promise(resolve => this._dataStream.write(packedBuffer, resolve));
315. this._lastWrite = writePromise;
316. await writePromise;
318. const loc = [offset, packedBuffer.length];
319. // indexes
320. if (this._indexManager) {
321. this._indexManager.insert(recToInsert, loc);
322. }
324. // Persist meta if we incremented nextId
325. if (this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER && !skipPrimaryKey && !hasPrimaryKeyValue) {
326. await this.persistMeta();
327. }
329. return this._primaryKey ? recToInsert[this._primaryKey] : recToInsert;
330. });
331. }
333. /**
334. * Update records matching a query.
335. * The callback receives the old record and must return the new record.
336. *
337. * @param {string|number|Function} mixed - Primary key value or query function
338. * @param {Function} callback - Receives old record, must return new record
339. * @param {Object} [options]
340. * @param {boolean} [options.upsert=false] - Insert if no records match
341. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads
342. * @returns {Promise<Object[]>} Array of updated records
343. *
344. * @example
345. * await db.update(0, record => {
346. * record.age = 32;
347. * delete record.address;
348. * return record;
349. * });
350. */
351. async update(mixed, callback, { upsert = false, index } = {}) {
352. if (typeof callback !== 'function') {
353. throw new Error('update requires a callback function as second argument');
354. }
355. let insertedRecords = await this.delete(mixed, {
356. index,
357. callback: async record => {
358. return this.insert(await callback(record), { skipPrimaryKey: true });
359. }
360. });
361. if (upsert && insertedRecords.length === 0) {
362. insertedRecords = await this.insert(await callback({}));
363. }
364. return insertedRecords;
365. }
367. /**
368. * Update records or insert if not found
369. *
370. * @param {string|number|Function} mixed - Primary key value or query function
371. * @param {Function} callback - Receives old record (or {} if inserting), must return new record
372. * @param {Object} [options]
373. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads
374. * @returns {Promise<Object[]>} Array of updated/inserted records
375. */
376. async upsert(mixed, callback, { index } = {}) {
377. return this.update(mixed, callback, { upsert: true, index });
378. }
380. /**
381. * Delete records matching a query
382. *
383. * @param {string|Function} mixed - Primary key value or query function
384. * @param {Object} [options]
385. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads
386. * @param {Function} [options.callback] - Internal callback per deleted record (used by update)
387. * @returns {Promise<Object[]>} Array of deleted records
388. *
389. * @example
390. * // Delete by primary key
391. * await db.delete(0);
392. *
393. * // Delete with query function
394. * await db.delete(r => r.age < 18);
395. *
396. * // Delete with index hint
397. * await db.delete(r => r.status === 'inactive', { index: { field: 'status', value: 'inactive' } });
398. */
399. async delete(mixed, { index, callback = async record => record } = {}) {
400. await this.init();
401. return this._withFileLock('delete', async () => {
402. const promises = [];
403. for await (const [record, offset] of this.find(mixed, { mode: 'mixed', index })) {
404. this._meta.deleted.push(offset);
405. if (this._indexManager) {
406. await this._indexManager.remove(record, this._primaryKey);
407. }
408. promises.push(callback(record));
409. }
410. await this.persistMeta();
411. return Promise.all(promises);
412. });
413. }
415. /**
416. * Finds records in the database
417. * @param {undefined|string|number|function} [mixed] - Query: undefined/null for all, primary key value for PK lookup, function for filter
418. * @param {Object} [options] - Query options
419. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads before filtering
420. * @returns {Cursor} A cursor for iterating over results
421. * @example
422. * // All records
423. * for await (const user of db.find()) { ... }
424. *
425. * // Primary key lookup
426. * const [user] = await db.find(68);
427. *
428. * // Filter with index range — only reads records in x 100+
429. * for await (const doc of db.find(r => r.x <= 200, { index: { field: 'x', from: 100 } })) { ... }
430. *
431. * // Index intersection — intersects offsets first, then streams matches
432. * for await (const doc of db.find(r => r.x <= 200, {
433. * index: [
434. * { field: 'x', from: 100 },
435. * { field: 'status', value: 'active' }
436. * ]
437. * })) { ... }
438. */
439. find(mixed, options = {}) {
440. if (typeof mixed === 'undefined' || mixed === null || typeof mixed === 'function') {
441. return new Cursor(this, mixed, options);
442. } else {
443. if (!this._primaryKey) {
444. throw new Error('No primary key specified.');
445. }
446. // Use indexed lookup when available
447. if (this._indexManager) {
448. return new Cursor(this, null, {
449. ...options,
450. _indexLookup: { field: this._primaryKey, value: mixed }
451. });
452. }
453. return new Cursor(this, record => {
454. return record[this._primaryKey] === mixed;
455. }, options);
456. }
457. }
459. /**
460. * Find records within a bounding box defined by 4 corners.
461. * Requires numeric indexes on x and y fields.
462. * Corners can be in any order — min/max are extracted automatically.
463. *
464. * @param {Array<{x: number, y: number}>} corners - 4 corner coordinates
465. * @param {function} [filter] - Optional additional filter function
466. * @returns {Cursor}
467. * @example
468. * const results = await db.boundingBox([
469. * { x: -5, y: 5 }, { x: 5, y: 5 },
470. * { x: 5, y: -5 }, { x: -5, y: -5 }
471. * ]);
472. */
473. boundingBox(corners, filter) {
474. const xs = corners.map(c => c.x);
475. const ys = corners.map(c => c.y);
476. const minX = Math.min(...xs), maxX = Math.max(...xs);
477. const minY = Math.min(...ys), maxY = Math.max(...ys);
479. return this.find(filter || null, {
480. index: [
481. { field: 'x', from: minX, to: maxX },
482. { field: 'y', from: minY, to: maxY }
483. ]
484. });
485. }
487. /**
488. * Execute a callback while holding the database lock.
489. * The lock is re-entrant: find/insert/delete/update called inside
490. * the callback reuse the same lock instead of deadlocking.
491. *
492. * Use this for compound operations that must be atomic, e.g.
493. * find-then-insert (login pattern).
494. *
495. * @param {Function} callback - Async function to execute under lock
496. * @returns {Promise<any>} The return value of the callback
497. *
498. * @example
499. * const user = await db.withLock(async () => {
500. * const [existing] = await db.find(u => u.email === email);
501. * if (existing) return existing;
502. * return db.insert({ email });
503. * });
504. */
505. async withLock(callback) {
506. await this.init();
507. return this._withFileLock('withLock', callback);
508. }
510. /**
511. * Generator that yields records from the database file
512. * @param {function|null} [queryFn=null] - Optional filter function
513. * @param {Object} [options] - Generator options
514. * @param {string} [options.mode='record'] - Mode: 'record', 'raw', 'offset', or 'mixed'
515. * @yields {Object|Buffer|Array} Records, buffers, or [record, offset, size] tuples depending on mode
516. */
517. async *recordGenerator(queryFn = null, { mode = 'record', _indexLookup, index } = {}) {
518. await this.init();
519. await this.refresh();
521. // Indexed primary key lookup — O(log n) instead of full scan
522. if (_indexLookup && this._indexManager) {
523. yield* this._indexedLookup(_indexLookup.field, _indexLookup.value, mode);
524. return;
525. }
527. // Index-based find: collect offsets from index(es), then stream only those records
528. if (index && this._indexManager) {
529. yield* this._indexedStream(index, queryFn, mode);
530. return;
531. }
533. // Flush pending writes so reads see all inserted data.
534. // (Waiting for 'drain' here could hang forever — 'drain' only fires
535. // after a write() returned false, which small writes never do.)
536. if (this._lastWrite) {
537. await this._lastWrite;
538. }
540. // Check if the data file exists before attempting to read it
541. try {
542. await stat(this._dataPath);
543. } catch (error) {
544. // File doesn't exist - return empty generator (no records)
545. return;
546. }
548. // Convert deleted array to Set for O(1) lookup instead of O(n)
549. const deletedSet = new Set(this._meta.deleted);
551. const readStream = createReadStream(this._dataPath);
552. let chunks = [];
553. let totalLength = 0;
554. let processedBytes = 0;
556. for await (const chunk of readStream) {
557. chunks.push(chunk);
558. totalLength += chunk.length;
560. while (true) {
561. // Exit 1: Not enough data to even read the 4-byte size header.
562. if (totalLength < 4) {
563. break;
564. }
566. // Safely read the header, even if it's split across chunks
567. let headerBuffer;
568. if (chunks[0].length >= 4) {
569. headerBuffer = chunks[0];
570. } else {
571. // The header is fragmented, so we must concat just enough to read it.
572. headerBuffer = Buffer.concat(chunks, 4);
573. }
574. const recSize = headerBuffer.readInt32LE(0);
576. // Validate the record size to prevent infinite loops
577. // A record must be at least as large as its header (4 bytes).
578. // A size of 0 or less is invalid and indicates corruption.
579. if (recSize <= 4) {
580. throw new Error(`Invalid record size read from stream: ${recSize}`);
581. }
583. // Exit 2: We have the size, but not the full record yet.
584. if (totalLength < recSize) {
585. break;
586. }
588. // skip deleted records (only after we have the full record)
589. if (deletedSet.has(processedBytes)) {
590. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
591. continue; // Goes back to while (true)
592. }
594. switch (mode) {
595. case 'raw':
596. yield Buffer.concat(chunks, recSize).subarray(0, recSize);
597. break;
599. case 'mixed':
600. case 'record':
601. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
602. // Skip the 4-byte size header
603. const data = deserialize(recBuffer.subarray(4));
604. const rec = this._classToUse
605. ? Object.assign(new this._classToUse(), data)
606. : data;
607. if (queryFn) {
608. if (queryFn(rec)) {
609. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
610. }
611. } else {
612. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
613. }
614. break;
616. case 'offset':
617. // if someone uses a queryFn with offset, we have to deserialize it first to perform the queryFn
618. if (queryFn) {
619. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
620. const rec = deserialize(recBuffer.subarray(4));
621. if (queryFn(rec)) {
622. yield [processedBytes, recSize];
623. }
624. } else {
625. yield [processedBytes, recSize];
626. }
627. break;
629. default:
630. throw new Error(`Invalid mode: ${mode}`);
631. }
633. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
634. }
635. }
636. }
638. /**
639. * Indexed lookup — reads records directly by offset from the index.
640. * Uses binary search on the index file for O(log n) lookups.
641. * @private
642. */
643. async *_indexedLookup(field, value, mode = 'record') {
644. const entries = await this._indexManager.get(field, value);
645. if (entries.length === 0) return;
647. const deletedSet = new Set(this._meta.deleted);
648. const fileHandle = await open(this._dataPath, 'r');
650. try {
651. for (const entry of entries) {
652. const [offset, length] = entry.loc;
653. if (deletedSet.has(offset)) continue;
655. const buffer = Buffer.alloc(length);
656. await fileHandle.read(buffer, 0, length, offset);
658. switch (mode) {
659. case 'raw':
660. yield buffer;
661. break;
662. case 'mixed': {
663. const data = deserialize(buffer.subarray(4));
664. const rec = this._classToUse
665. ? Object.assign(new this._classToUse(), data)
666. : data;
667. yield [rec, offset, length];
668. break;
669. }
670. case 'record':
671. default: {
672. const data = deserialize(buffer.subarray(4));
673. const rec = this._classToUse
674. ? Object.assign(new this._classToUse(), data)
675. : data;
676. yield rec;
677. break;
678. }
679. }
680. }
681. } finally {
682. await fileHandle.close();
683. }
684. }
686. /**
687. * Collect offsets from one or more indexes, optionally intersect, then stream records.
688. * @param {Object|Array} index - Single index hint or array of hints
689. * @param {function|null} queryFn - Optional filter function
690. * @param {string} mode - Output mode: 'record', 'raw', or 'mixed'
691. * @private
692. */
693. async *_indexedStream(index, queryFn, mode = 'record') {
694. const hints = Array.isArray(index) ? index : [index];
695. const deletedSet = new Set(this._meta.deleted);
697. // Collect offset sets from each index hint
698. const offsetSets = [];
699. for (const hint of hints) {
700. const offsets = new Map(); // offset → [offset, length]
701. if (hint.value !== undefined) {
702. // Exact match via get()
703. const entries = await this._indexManager.get(hint.field, hint.value);
704. for (const e of entries) {
705. if (!deletedSet.has(e.loc[0])) offsets.set(e.loc[0], e.loc);
706. }
707. } else {
708. // Range scan via entries()
709. for await (const e of this._indexManager.entries(hint.field, { from: hint.from, to: hint.to, direction: hint.direction })) {
710. if (!deletedSet.has(e.loc[0])) offsets.set(e.loc[0], e.loc);
711. }
712. }
713. offsetSets.push(offsets);
714. }
716. // Intersect: keep only offsets present in ALL sets
717. let locations;
718. if (offsetSets.length === 1) {
719. locations = Array.from(offsetSets[0].values());
720. } else {
721. // Start with smallest set for efficiency
722. offsetSets.sort((a, b) => a.size - b.size);
723. const [smallest, ...rest] = offsetSets;
724. locations = [];
725. for (const [offset, loc] of smallest) {
726. if (rest.every(s => s.has(offset))) locations.push(loc);
727. }
728. }
730. // Stream records from the intersected locations
731. const fileHandle = await open(this._dataPath, 'r');
732. try {
733. for (const [offset, length] of locations) {
734. const buffer = Buffer.alloc(length);
735. await fileHandle.read(buffer, 0, length, offset);
737. if (mode === 'raw') {
738. yield buffer;
739. } else {
740. const data = deserialize(buffer.subarray(4));
741. const rec = this._classToUse
742. ? Object.assign(new this._classToUse(), data)
743. : data;
745. if (queryFn && !queryFn(rec)) continue;
747. if (mode === 'mixed') {
748. yield [rec, offset, length];
749. } else {
750. yield rec;
751. }
752. }
753. }
754. } finally {
755. await fileHandle.close();
756. }
757. }
759. /**
760. * Pick up changes made by OTHER processes. Called automatically before every
761. * read operation and after acquiring the write lock.
762. *
763. * The in-memory meta is authoritative for this process — it may hold
764. * un-persisted mutations of an in-flight write. It is only replaced when the
765. * on-disk copy carries a NEWER version, i.e. another process persisted under
766. * the file lock. (Unconditionally reloading here was the root cause of the
767. * lost-tombstone/lost-insert corruption under concurrent reads + writes.)
768. *
769. * Also detects external data file changes: appended records are indexed via
770. * a tail scan, a replaced file (external compaction) triggers a full reopen.
771. */
772. async refresh() {
773. try {
774. const diskMeta = JSON.parse(await readFile(this._metaPath));
775. if ((diskMeta.version || 0) > (this._meta.version || 0)) {
776. this._meta = { nextId: 0, deleted: [], ...diskMeta };
777. }
778. } catch (e) {
779. // file missing or torn write in progress — keep current meta
780. }
782. const dataStat = await stat(this._dataPath).catch(() => null);
783. if (dataStat) {
784. if (this._dataIno !== null && dataStat.ino !== this._dataIno) {
785. await this._reopenAfterExternalReplace(dataStat);
786. } else if (this._indexManager) {
787. await this._indexManager.catchUp(this._dataPath, dataStat.size);
788. }
789. }
790. }
792. /**
793. * The data file was replaced by another process (external compaction):
794. * all offsets changed and our write stream points at the orphaned inode.
795. * Reopen the stream, force-reload meta, and reset the index state.
796. * @private
797. */
798. async _reopenAfterExternalReplace(dataStat) {
799. this.debug(`Data file was replaced externally (compaction), reopening ${this._dataPath}`);
800. if (this._dataStream) {
801. await new Promise((resolve, reject) => {
802. this._dataStream.end((err) => err ? reject(err) : resolve());
803. });
804. this._dataStream = createWriteStream(this._dataPath, { flags: 'a' });
805. }
806. this._dataIno = dataStat.ino;
807. try {
808. this._meta = { nextId: 0, deleted: [], ...JSON.parse(await readFile(this._metaPath)) };
809. } catch (e) { }
810. if (this._indexManager) {
811. await this._indexManager.resetAfterReplace(this._dataPath);
812. }
813. }
815. async persistMeta() {
816. // Monotonic version: refresh() in other processes reloads only when it
817. // sees a version newer than its in-memory one
818. this._meta.version = (this._meta.version || 0) + 1;
820. const metaToSave = { ...this._meta };
822. // Only save nextId if we have a numeric primary key
823. if (this._primaryKeyType !== PrimaryKeyType.NUMBER) {
824. delete metaToSave.nextId;
825. }
827. if (this._schema) {
828. this._meta.schema = this._schema;
829. metaToSave.schema = this._schema;
830. }
832. // Atomic write (tmp + rename) so concurrent readers never parse a torn file
833. const tmpPath = `${this._metaPath}.${process.pid}.tmp`;
834. await writeFile(tmpPath, JSON.stringify(metaToSave));
835. await rename(tmpPath, this._metaPath);
836. }
838. /**
839. * Compact the database by removing deleted records
840. * This rewrites the data file without tombstones
841. *
842. * @returns {Promise<void>}
843. */
844. async compact({ duringInit = false } = {}) {
845. return this._withFileLock('compact', async () => {
846. const writeStream = createWriteStream(this._dataPath + '.tmp', { flags: 'w' });
848. const records = duringInit
849. ? this._rawRecordsForCompaction()
850. : this.find(null, { mode: 'raw' });
851. for await (const binary of records) {
852. writeStream.write(binary);
853. }
855. await new Promise((resolve, reject) => {
856. writeStream.end((err) => err ? reject(err) : resolve());
857. });
859. // The old write stream points at the inode that rename() is about to
860. // orphan — writes to it would be silently lost. Close it first.
861. if (this._dataStream) {
862. await new Promise((resolve, reject) => {
863. this._dataStream.end((err) => err ? reject(err) : resolve());
864. });
865. }
867. // rename is atomic, so we can just rename the file and it will replace the old one
868. await rename(this._dataPath + '.tmp', this._dataPath);
870. if (this._dataStream) {
871. this._dataStream = createWriteStream(this._dataPath, { flags: 'a' });
872. }
873. this._dataIno = (await stat(this._dataPath).catch(() => null))?.ino ?? null;
875. this._meta.deleted = [];
876. await this.persistMeta();
878. // Rebuild indexes — offsets changed after compaction
879. if (!duringInit && this._indexManager) {
880. await this._indexManager.init(this._dataPath, { forceRebuild: true });
881. }
882. });
883. }
885. async *_rawRecordsForCompaction() {
886. const deletedSet = new Set(this._meta.deleted);
887. try {
888. await stat(this._dataPath);
889. } catch (error) {
890. return;
891. }
893. const readStream = createReadStream(this._dataPath);
894. let chunks = [];
895. let totalLength = 0;
896. let processedBytes = 0;
898. for await (const chunk of readStream) {
899. chunks.push(chunk);
900. totalLength += chunk.length;
902. while (true) {
903. if (totalLength < 4) break;
905. const headerBuffer = chunks[0].length >= 4
906. ? chunks[0]
907. : Buffer.concat(chunks, 4);
908. const recSize = headerBuffer.readInt32LE(0);
909. if (recSize <= 4) {
910. throw new Error(`Invalid record size read from stream: ${recSize}`);
911. }
912. if (totalLength < recSize) break;
914. const record = Buffer.concat(chunks, recSize).subarray(0, recSize);
915. if (!deletedSet.has(processedBytes)) {
916. yield record;
917. }
919. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
920. }
921. }
922. }
924. /**
925. * Run a function exclusively: serialized against other operations in this
926. * process (FIFO queue) and against other processes (lock file).
927. *
928. * Re-entrant only for operations called INSIDE the callback's async call
929. * chain (find inside delete, insert inside withLock, ...). Operations
930. * started concurrently from outside wait their turn.
931. *
932. * On acquiring the file lock, the on-disk meta is refreshed so writes in
933. * other processes (fresh nextId, tombstones) are visible before mutating.
934. * @private
935. */
936. async _withFileLock(operation, fn) {
937. if (lockContext.getStore()?.owner === this) {
938. // Causal re-entry: our call chain already holds this db's lock
939. return fn();
940. }
942. // In-process FIFO queue — avoids the 100ms lock file polling between
943. // concurrent operations of the same instance
944. const prev = this._mutexTail;
945. let releaseQueue;
946. this._mutexTail = new Promise(r => releaseQueue = r);
947. await prev;
949. try {
950. await this._acquireFileLock(operation);
951. try {
952. // Sync with other processes' writes before entering the critical
953. // section — also during init (paths are set before any locked op)
954. if (this._metaPath) await this.refresh();
955. return await lockContext.run({ owner: this }, fn);
956. } finally {
957. await unlink(this._lockPath).catch(() => { });
958. }
959. } finally {
960. releaseQueue();
961. }
962. }
964. async _acquireFileLock(operation, retries = 0) {
965. while (true) {
966. try {
967. await writeFile(this._lockPath, String(process.pid), { flag: 'wx' });
968. return;
969. } catch (e) {
970. if (e.code !== 'EEXIST') throw e;
972. // Stale lock recovery: a crashed holder never unlinks its lock file.
973. // If the lock file is older than staleLockTimeout, take it over.
974. if (this._staleLockTimeout > 0) {
975. const lockStat = await stat(this._lockPath).catch(() => null);
976. if (lockStat && Date.now() - lockStat.mtimeMs > this._staleLockTimeout) {
977. const stalePath = `${this._lockPath}.stale-${process.pid}`;
978. try {
979. await rename(this._lockPath, stalePath);
980. const staleStat = await stat(stalePath);
981. if (Date.now() - staleStat.mtimeMs > this._staleLockTimeout) {
982. console.warn(`[mpackdb] removed stale lock ${this._lockPath} (held > ${this._staleLockTimeout}ms, holder presumed dead)`);
983. await unlink(stalePath).catch(() => { });
984. } else {
985. // raced a fresh lock — put it back
986. await rename(stalePath, this._lockPath).catch(() => { });
987. }
988. } catch (e2) {
989. // another waiter beat us to the takeover
990. }
991. continue;
992. }
993. }
995. this.debug(`Waiting for lock file ${operation} retry #${retries}`);
996. await new Promise(resolve => setTimeout(resolve, 25));
997. retries++;
998. }
999. }
1000. }
1002. /**
1003. * Retrieves a document by its file offset and length
1004. * @param {Array} location - [offset, length] tuple
1005. * @returns {Promise<Object|null>} The deserialized document or null
1006. * @private
1007. * @deprecated Currently unused - may be removed in future versions
1008. */
1009. async _getDocByLocation([offset, length]) {
1010. if (!offset || length === 0) return null;
1011. const fileHandle = await open(this._dataPath, 'r');
1012. try {
1013. const buffer = Buffer.alloc(length);
1014. await fileHandle.read(buffer, 0, length, offset);
1015. return deserialize(buffer);
1016. } finally {
1017. await fileHandle.close();
1018. }
1019. }
1021. /**
1022. * Close the database and persist all pending changes
1023. * Should be called before process exit
1024. *
1025. * @returns {Promise<void>}
1026. */
1027. async close() {
1028. // Persist indexes before closing
1029. if (this._indexManager) {
1030. await this._indexManager.close();
1031. }
1033. // Close data stream
1034. if (this._dataStream) {
1035. await new Promise((resolve, reject) => {
1036. this._dataStream.end((err) => err ? reject(err) : resolve());
1037. });
1038. }
1040. // Remove signal handlers
1041. if (this._processExitHandler) {
1042. process.off('SIGINT', this._processExitHandler);
1043. process.off('SIGTERM', this._processExitHandler);
1044. this._processExitHandler = null;
1045. }
1046. }
1048. debug(...args) {
1049. if (this._debug) {
1050. console.log(...args);
1051. }
1052. }
1053. }
1055. /**
1056. * Helper function to remove a processed chunk from the buffer
1057. * @param {number} recSize - Size of the record to remove
1058. * @param {number} processedBytes - Current offset in the file
1059. * @param {number} totalLength - Total length of buffered data
1060. * @param {Buffer[]} chunks - Array of buffer chunks
1061. * @returns {[number, number]} Updated [processedBytes, totalLength]
1062. */
1063. function removeChunk(recSize, processedBytes, totalLength, chunks) {
1064. processedBytes += recSize;
1065. totalLength -= recSize;
1067. let bytesToRemove = recSize;
1068. while (bytesToRemove > 0 && chunks.length > 0) {
1069. const currentChunk = chunks[0];
1070. if (bytesToRemove >= currentChunk.length) {
1071. bytesToRemove -= currentChunk.length;
1072. chunks.shift();
1073. } else {
1074. chunks[0] = currentChunk.subarray(bytesToRemove);
1075. bytesToRemove = 0;
1076. }
1077. }
1079. return [processedBytes, totalLength];
1080. }
1082. export default MPackDB;