Merge remote-tracking branch 'origin/main' into vm-load-plugins

Author: kriszyp

resources/RecordEncoder.ts

@@ -75,6 +75,7 @@ export let lastMetadata: Entry | null = null;
 export class RecordEncoder extends Encoder {
 	structureUpdate?: any;
 	isRocksDB: boolean;
+	name: string;
 	constructor(options) {
 		options.useBigIntExtension = true;
 		/**
@@ -116,14 +117,23 @@ export class RecordEncoder extends Encoder {
 					if (expiresAt >= 0) {
 						valueStart += 8; // make room for expiration timestamp
 						expiresAtNextEncoding = -1; // reset indicator to mean no expiration
+						if (!(metadata & HAS_EXPIRATION)) {
+							throw new Error('Expiration included, but not in metadata flags');
+						}
 					}
 					if (residencyId) {
 						valueStart += 4; // make room for residency id
 						residencyIdAtNextEncoding = 0; // reset indicator to mean no residency id
+						if (!(metadata & HAS_RESIDENCY_ID)) {
+							throw new Error('Residency id included, but not in metadata flags');
+						}
 					}
 					if (nodeId >= 0) {
 						valueStart += 4; // make room for node id
 						nodeIdAtNextEncoding = -1; // reset indicator to mean no node id
+						if (!(metadata & HAS_NODE_ID)) {
+							throw new Error('Node id included, but not in metadata flags');
+						}
 					}
 					if (additionalAuditRefs && additionalAuditRefs.length > 0) {
 						valueStart += 1 + additionalAuditRefs.length * 12; // 1 byte for count + 8 bytes version + 4 bytes nodeId per ref
@@ -553,17 +563,17 @@ export function recordUpdater(store, tableId, auditStore) {
 				residencyIdAtNextEncoding = residencyId;
 				metadataInNextEncoding |= HAS_RESIDENCY_ID;
 				extendedType |= HAS_CURRENT_RESIDENCY_ID;
-			} // else residencyIdAtNextEncoding = 0;
+			} else residencyIdAtNextEncoding = 0;
 			const nodeId = options?.nodeId;
 			if (nodeId >= 0) {
 				nodeIdAtNextEncoding = nodeId;
 				metadataInNextEncoding |= HAS_NODE_ID;
-			} // else nodeIdAtNextEncoding = -1;
+			} else nodeIdAtNextEncoding = -1;
 			const additionalAuditRefs = options?.additionalAuditRefs;
 			if (additionalAuditRefs && additionalAuditRefs.length > 0) {
 				additionalAuditRefsNextEncoding = additionalAuditRefs;
 				metadataInNextEncoding |= HAS_ADDITIONAL_AUDIT_REFS;
-			} // else additionalAuditRefsNextEncoding = undefined;
+			} else additionalAuditRefsNextEncoding = undefined;
 			const previousAdditionalAuditRefs = existingEntry?.additionalAuditRefs;
 			if (previousAdditionalAuditRefs && previousAdditionalAuditRefs.length > 0) {
 				extendedType |= HAS_ADDITIONAL_AUDIT_REFS_AUDIT;
@@ -606,7 +616,6 @@ export function recordUpdater(store, tableId, auditStore) {
 				const structureVersion = store.encoder.structures.length + (store.encoder.typedStructs?.length ?? 0);
 				const nodeId = options?.nodeId ?? server.replication?.getThisNodeId(auditStore) ?? 0;
 				const viaNodeId = options?.viaNodeId ?? nodeId;
-				logger.debug('recording audit entry', { id, newVersion, previousVersion: existingEntry?.version, nodeId });
 				if (resolveRecord && existingEntry?.localTime) {
 					const replacingId = existingEntry?.localTime;
 					const replacingEntry = auditStore.get(replacingId, tableId, id);

resources/RocksTransactionLogStore.ts

@@ -317,6 +317,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 		let totalSize = 0;
 		const logs = [];
 		for (const log of this.loadLogs()) {
+			if (!log) continue;
 			const size = log.getLogFileSize();
 			totalSize += size;
 			logs.push({ name: log.name, size });

resources/databases.ts

@@ -113,6 +113,7 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo
 		db = RocksDatabase.open(new RocksIndexStore(path, options)) as RocksDatabaseEx;
 	} else {
 		db = RocksDatabase.open(path, options) as RocksDatabaseEx;
+		db.encoder.name = options.name;
 	}
 	db.env = {};
 	return db;
@@ -847,7 +848,7 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 		if (attribute.expiresAt) attribute.indexed = true;
 	}
 	let hasChanges;
-	let txnCommit;
+	let releaseExclusiveLock: () => void;
 	if (Table) {
 		primaryKey = Table.primaryKey;
 		if (Table.primaryStore.rootStore.status === 'closed') {
@@ -891,10 +892,10 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 			attributesDbi = rootStore.dbisDb = rootStore.openDB(INTERNAL_DBIS_NAME, internalDbiInit);
 		}
 
-		startTxn(); // get an exclusive lock on the database so we can verify that we are the only thread creating the table (and assigning the table id)
+		exclusiveLock(); // get an exclusive lock on the database so we can verify that we are the only thread creating the table (and assigning the table id)
 		if (attributesDbi.getSync(dbiName)) {
 			// table was created while we were setting up
-			if (txnCommit) txnCommit();
+			if (releaseExclusiveLock) releaseExclusiveLock();
 			resetDatabases();
 			return table(tableDefinition);
 		}
@@ -969,7 +970,7 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 		const attribute = attributes.find((attribute) => attribute.name === attribute_name);
 		const removeIndex = !attribute?.indexed && value.indexed && !value.isPrimaryKey;
 		if (!attribute || removeIndex) {
-			startTxn();
+			exclusiveLock();
 			hasChanges = true;
 			if (!attribute) attributesDbi.remove(key);
 			if (removeIndex) {
@@ -1012,7 +1013,7 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 					if (replicate !== undefined) updatedPrimaryAttribute.replicate = replicate;
 					if (attribute.type) updatedPrimaryAttribute.type = attribute.type;
 					hasChanges = true; // send out notification of the change
-					startTxn();
+					exclusiveLock();
 					attributesDbi.put(dbiKey, updatedPrimaryAttribute);
 				}
 
@@ -1038,7 +1039,7 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 					attributeDescriptor.restartNumber < workerData?.restartNumber
 				) {
 					hasChanges = true;
-					startTxn();
+					exclusiveLock();
 					attributeDescriptor = attributesDbi.getSync(dbiKey);
 					if (
 						changed ||
@@ -1068,12 +1069,12 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 				indices[attribute.name] = dbi;
 			} else if (changed) {
 				hasChanges = true;
-				startTxn();
+				exclusiveLock();
 				attributesDbi.put(dbiKey, attribute);
 			}
 		}
 	} finally {
-		if (txnCommit) txnCommit();
+		if (releaseExclusiveLock) releaseExclusiveLock();
 	}
 	if (hasChanges) {
 		Table.schemaVersion++;
@@ -1102,15 +1103,24 @@ export function table<TableResourceType>(tableDefinition: TableDefinition): Tabl
 	logger.trace(`${tableName} table loaded`);
 
 	return Table as TableResourceType;
-	function startTxn() {
-		if (txnCommit) return;
-		rootStore.transactionSync(() => {
-			return {
-				then(callback) {
-					txnCommit = callback;
-				},
+	// Acquire an exclusive lock for attribute updates
+	function exclusiveLock() {
+		if (releaseExclusiveLock) return;
+		if (rootStore instanceof RocksDatabase) {
+			while (!rootStore.tryLock('update-attributes')) {} // use a spin lock, we really need an synchronous exclusive lock here
+			releaseExclusiveLock = () => {
+				rootStore.unlock('update-attributes');
 			};
-		});
+		} else {
+			// we only need an exclusive transaction lock in lmdb
+			rootStore.transactionSync(() => {
+				return {
+					then(callback) {
+						releaseExclusiveLock = callback;
+					},
+				};
+			});
+		}
 	}
 }
 const MAX_OUTSTANDING_INDEXING = 1000;

utility/lmdb/environmentUtility.js

@@ -11,7 +11,7 @@ const { OpenDBIObject } = require('./OpenDBIObject.js');
 const OpenEnvironmentObject = require('./OpenEnvironmentObject.js');
 const lmdbTerms = require('./terms.js');
 const hdbTerms = require('../hdbTerms.ts');
-const { table, resetDatabases } = require('../../resources/databases.ts');
+const { resetDatabases } = require('../../resources/databases.ts');
 const envMngr = require('../environment/environmentManager.js');
 
 const INTERNAL_DBIS_NAME = lmdbTerms.INTERNAL_DBIS_NAME;
@@ -21,41 +21,6 @@ const MDB_LEGACY_LOCK_FILE_NAME = 'lock.mdb';
 const MDB_FILE_EXTENSION = '.mdb';
 const MDB_LOCK_FILE_SUFFIX = '-lock';
 
-/**
- * This class is used to create the transaction & cursor objects needed to perform search on a dbi as well as a function to close both objects after use
- */
-class TransactionCursor {
-	/**
-	 * create the TransactionCursor object
-	 * @param {lmdb.RootDatabase} env - environment object to create the transaction & cursor from
-	 * @param {String} attribute - name of the attribute to create the cursor against
-	 * @param {Boolean} [writeCursor] - optional, dictates if the cursor created will be a readOnly cursor or not
-	 */
-	constructor(env, attribute, writeCursor = false) {
-		this.dbi = openDBI(env, attribute);
-		this.key_type = this.dbi[lmdbTerms.DBI_DEFINITION_NAME].key_type;
-		this.isPrimaryKey = this.dbi[lmdbTerms.DBI_DEFINITION_NAME].isPrimaryKey;
-		this.txn = env.beginTxn({ readOnly: writeCursor === false });
-		this.cursor = new lmdb.Cursor(this.txn, this.dbi);
-	}
-
-	/**
-	 * function to close the read cursor & abort the transaction
-	 */
-	close() {
-		this.cursor.close();
-		this.txn.abort();
-	}
-
-	/**
-	 * function to close the read cursor & abort the transaction
-	 */
-	commit() {
-		this.cursor.close();
-		this.txn.commit();
-	}
-}
-
 /***  VALIDATION FUNCTIONS ***/
 
 /**
@@ -172,18 +137,6 @@ async function createEnvironment(basePath, envName, isTxn = false, isV3 = false)
 	}
 }
 
-async function copyEnvironment(basePath, envName, _destinationPath, _compactEnvironment = true) {
-	pathEnvNameValidation(basePath, envName);
-	envName = envName.toString();
-	let environmentPath = path.join(basePath, envName);
-	return table({
-		table: envName,
-		database: path.parse(basePath).name,
-		path: environmentPath,
-		attributes: [{ name: 'id', isPrimaryKey: true }],
-	});
-}
-
 /**
  * opens an environment
  * @returns {lmdb.RootDatabase} - lmdb environment object
@@ -456,22 +409,6 @@ function statDBI(env, dbiName) {
 	return stats;
 }
 
-/**
- * gets the byte size of an environment file
- * @param {String} environmentBasePath
- * @param {String} tableName
- * @returns {Promise<number>}
- */
-async function environmentDataSize(environmentBasePath, tableName) {
-	try {
-		let environmentPath = path.join(environmentBasePath, tableName + MDB_FILE_EXTENSION);
-		let statResult = await fs.stat(environmentPath);
-		return statResult['size'];
-	} catch {
-		throw new Error(LMDB_ERRORS.INVALID_ENVIRONMENT);
-	}
-}
-
 /**
  * removes a named database from an environment
  * @param {lmdb.RootDatabase} env - environment object used thigh level to interact with all data in an environment
@@ -536,8 +473,5 @@ module.exports = {
 	statDBI,
 	deleteEnvironment,
 	initializeDBIs,
-	TransactionCursor,
-	environmentDataSize,
-	copyEnvironment,
 	closeEnvironment,
 };

utility/lmdb/searchUtility.js

@@ -829,7 +829,6 @@ function batchHashSearch(transactionOrEnv, hash_attribute, fetchAttributes, ids,
  * @param {Array.<String>} fetchAttributes - string array of attributes to pull from the object
  * @param {Array.<String>} ids - list of ids to search
  * @param {[]} [_notFound] -optional,  meant to be an array passed by reference so that skipped ids can be aggregated.
- * @returns {TransactionCursor}
  */
 function initializeBatchSearchByHash(transactionOrEnv, hash_attribute, fetchAttributes, ids, _notFound) {
 	common.validateEnv(transactionOrEnv);