Added concurrency-safety for multi-operation transactions

Author: keijokapp

README.md

@@ -68,7 +68,6 @@ The adapter is subject to the fundamental [limitations](https://apple.github.io/
 Also, there are non-FoundationDB-specific limitations:
 
 - All write operations update "last update seq" and "doc count" keys, so concurrent write transactions are effectively guaranteed to conflict and wouldn't benefit from concurrency. This can also affect read operations, eg if they need to update views.
-- Concurrent PouchDB actions within a single transaction are not isolated. Doing any write operations in parallel with other operations results in an undefined behavior.
 
 Some of these limitations could be easily solved. If these limitations are a significant problem for you, please create an issue.
 

lib/adapter.js

@@ -2,6 +2,7 @@
 
 import assert from 'assert';
 import crypto from 'crypto';
+import Deque from 'double-ended-queue';
 import { clone, filterChange, uuid } from 'pouchdb-utils';
 import {
 	collectConflicts,
@@ -131,13 +132,90 @@ function callbackify(fn) {
 	};
 }
 
+/**
+ * @param {fdb.Transaction} db
+ * @param {TransactionState} state
+ */
+async function executeQueue(db, state) {
+	const { queue } = state;
+
+	while (queue.length > 0) {
+		const task = /** @type {Task} */(queue.peekFront());
+
+		if (!task.write) {
+			const readTasks = [task];
+
+			for (let i = 1; i < queue.length; i++) {
+				const nextTask = /** @type {Task} */(queue.get(i));
+
+				if (nextTask.write) {
+					break;
+				}
+
+				readTasks.push(nextTask);
+			}
+
+			await Promise.all(readTasks.map(
+				({ fn, resolve, reject }) => fn(db).then(resolve, reject)
+			));
+
+			readTasks.forEach(() => {
+				queue.shift();
+			});
+		} else {
+			const { fn, resolve, reject } = task;
+
+			await fn(db).then(resolve, reject);
+
+			queue.shift();
+		}
+	}
+}
+
 /**
  * @typedef {(
  *   | fdb.Database
  *   | fdb.Transaction
  * )} Actionable
+ * @typedef {{ queue: Deque<Task> }} TransactionState
+ * @typedef {{
+ *   fn: (tn: fdb.Transaction) => Promise<unknown>,
+ *   resolve: (value: unknown) => void,
+ *   reject: (reason?: any) => void,
+ *   write: boolean
+ * }} Task
  */
 
+/**
+ * A quasi-global state needed for locks for when concurrent PouchDB operations are made with the
+ * same transaction. The key is the hidden context object of the transaction.
+ * @type {WeakMap<fdb.Transaction, Record<string, TransactionState>>}
+ */
+const transactionState = new WeakMap();
+
+/**
+ * @param {fdb.Transaction} tn
+ * @param {string} name
+ * @returns {TransactionState}
+ */
+function getTransactionState(tn, name) {
+	let state = transactionState.get(/** @type {any} */(tn)._ctx);
+
+	if (state == null) {
+		state = {};
+
+		transactionState.set(/** @type {any} */(tn)._ctx, state);
+	}
+
+	if (!(name in state)) {
+		state[name] = {
+			queue: new Deque()
+		};
+	}
+
+	return state[name];
+}
+
 /**
  * @param {any} api
  * @param {{ db: Actionable, name: string, revs_limit?: number }} opts
@@ -148,12 +226,46 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 	 * @param {(tn: fdb.Transaction) => Promise<T>} fn
 	 * @returns {Promise<T>}
 	 */
-	function doTn(fn) {
+	function doReadTn(fn) {
+		return doTn(fn, false);
+	}
+
+	/**
+	 * @template T
+	 * @param {(tn: fdb.Transaction) => Promise<T>} fn
+	 * @returns {Promise<T>}
+	 */
+	function doWriteTn(fn) {
+		return doTn(fn, true);
+	}
+
+	/**
+	 * @template T
+	 * @param {(tn: fdb.Transaction) => Promise<T>} fn
+	 * @param {boolean} write
+	 * @returns {Promise<T>}
+	 */
+	function doTn(fn, write) {
 		if ('doTn' in db) {
 			return db.doTn(fn);
 		}
 
-		return fn(db);
+		const state = getTransactionState(db, name);
+
+		const promise = new Promise((resolve, reject) => {
+			state.queue.push({
+				fn,
+				write,
+				resolve,
+				reject
+			});
+		});
+
+		if (state.queue.length === 1) {
+			executeQueue(db, state);
+		}
+
+		return promise;
 	}
 
 	const subspace = db.subspace.at(fdb.tuple.pack(name), defaultTransformer, defaultTransformer);
@@ -187,9 +299,11 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 
 	api._id = callbackify(
 		/**
+		 * TODO: That is concurrency-safe with most other read operations and, in most cases, with
+		 * itself so it might not be necessary registeer it as a write transaction.
 		 * @returns {Promise<string>}
 		 */
-		() => doTn(async tn => {
+		() => doWriteTn(async tn => {
 			let instanceId = await /** @type {Promise<undefined | string>} */(
 				tn.at(stores.metaStore).get(UUID_KEY)
 			);
@@ -208,7 +322,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		/**
 		 * @returns {Promise<{ doc_count: number, update_seq: number }>}
 		 */
-		async () => doTn(async tn => {
+		async () => doReadTn(async tn => {
 			const [docCount = 0, updateSeq = 0] = await Promise.all([
 				/** @type {Promise<undefined | number>} */(tn.at(stores.metaStore).get(DOC_COUNT_KEY)),
 				/** @type {Promise<undefined | number>} */(tn.at(stores.metaStore).get(UPDATE_SEQ_KEY))
@@ -226,7 +340,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * @param {import('./types.js').Id} id
 		 * @param {{ rev?: import('./types.js').Rev, latest?: boolean }} param
 		 */
-		(id, { rev, latest } = {}) => doTn(async tn => {
+		(id, { rev, latest } = {}) => doReadTn(async tn => {
 			const metadata = await tn.at(stores.docStore).get(id);
 
 			if (metadata == null) {
@@ -278,7 +392,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * @param {{ binary?: boolean }} param3
 		 * @returns {Promise<Buffer | string>}
 		 */
-		(docId, attachId, attachment, { binary } = {}) => doTn(async tn => {
+		(docId, attachId, attachment, { binary } = {}) => doReadTn(async tn => {
 			const { digest } = attachment;
 
 			const attach = await tn.at(stores.binaryStore).get(digest) ?? Buffer.allocUnsafe(0);
@@ -298,7 +412,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * @param {{ new_edits: boolean }} opts
 		 * @returns {Promise<import('./types.js').BulkDocsResultRow[]>}
 		 */
-		({ docs: userDocs }, opts) => doTn(async tn => {
+		({ docs: userDocs }, opts) => doWriteTn(async tn => {
 			const docInfos = userDocs.map(doc => {
 				if (isLocalDoc(doc)) {
 					return doc;
@@ -561,7 +675,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * }} opts
 		 * @returns {Promise<import('./types.js').AllDocsResult>}
 		 */
-		opts => doTn(async tn => {
+		opts => doReadTn(async tn => {
 			/** @type {[import('./types.js').AllDocsResultRow[], number, number]} */
 			const [rows, docCount = 0, updateSeq = 0] = await Promise.all([
 				getAllDocsRows(opts, tn),
@@ -876,7 +990,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 
 		for (;;) {
 			// eslint-disable-next-line no-loop-func
-			const { it, watch } = await doTn(async tn => {
+			const { it, watch } = await doReadTn(async tn => {
 				const it = await tn.at(stores.bySeqStore).getRangeAll(
 					lastSeq,
 					Infinity,
@@ -909,7 +1023,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 
 				if (!metadata) {
 					metadata = /** @type {import('./types.js').Metadata} */(
-						await doTn(tn => tn.at(stores.docStore).get(doc._id))
+						await doReadTn(tn => tn.at(stores.docStore).get(doc._id))
 					);
 
 					if (isLocalId(metadata.id)) {
@@ -930,7 +1044,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 
 				const winningDoc = winningRev === doc._rev
 					? doc
-					: await doTn(tn => tn.at(stores.bySeqStore).get(metadata.rev_map[winningRev]));
+					: await doReadTn(tn => tn.at(stores.bySeqStore).get(metadata.rev_map[winningRev]));
 
 				assert(winningDoc);
 
@@ -952,7 +1066,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 
 						// fetch attachment immediately for the benefit
 						// of live listeners
-						change.doc._attachments = await doTn(async tn => Object.fromEntries(
+						change.doc._attachments = await doReadTn(async tn => Object.fromEntries(
 							await Promise.all(Object.entries(attachments).map(
 								async ([fileName, att]) => [
 									fileName,
@@ -1018,7 +1132,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * @returns {Promise<import('./types.js').RevTreePath[]>}
 		 */
 		async docId => {
-			const metadata = await doTn(tn => tn.at(stores.docStore).get(docId));
+			const metadata = await doReadTn(tn => tn.at(stores.docStore).get(docId));
 
 			if (metadata == null) {
 				throw createError(MISSING_DOC);
@@ -1033,7 +1147,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * @param {import('./types.js').Id} docId
 		 * @param {import('./types.js').Rev[]} revs
 		 */
-		(docId, revs) => doTn(tn => doCompaction(docId, revs, tn))
+		(docId, revs) => doWriteTn(tn => doCompaction(docId, revs, tn))
 	);
 
 	/**
@@ -1129,7 +1243,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 * @returns {Promise<import('./types.js').LocalDoc>}
 		 */
 		async id => {
-			const value = await doTn(tn => tn.at(stores.localStore).get(id));
+			const value = await doReadTn(tn => tn.at(stores.localStore).get(id));
 
 			if (value == null) {
 				throw createError(MISSING_DOC);
@@ -1147,7 +1261,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 */
 		(doc, opts) => opts?.ctx
 			? putLocal(doc, opts.ctx)
-			: doTn(tn => putLocal(doc, tn))
+			: doWriteTn(tn => putLocal(doc, tn))
 	);
 
 	/**
@@ -1187,7 +1301,7 @@ export default function FoundationdbAdapter(api, { db, name, revs_limit: revLimi
 		 */
 		(doc, opts) => opts?.ctx
 			? removeLocal(doc, opts.ctx)
-			: doTn(tn => removeLocal(doc, tn))
+			: doWriteTn(tn => removeLocal(doc, tn))
 	);
 
 	/**

package-lock.json

@@ -16,13 +16,15 @@
     "lint": "eslint lib test eslint.config.js"
   },
   "dependencies": {
+    "double-ended-queue": "^2.1.0-0",
     "foundationdb": "^2.0.1",
     "pouchdb-errors": "^8.0.1",
     "pouchdb-merge": "^8.0.1",
     "pouchdb-utils": "^8.0.1"
   },
   "devDependencies": {
     "@arbendium/eslint-config-base": "^0.2.1",
+    "@types/double-ended-queue": "^2.1.7",
     "@types/node": "^20.12.7",
     "eslint": "^8.54.0",
     "express": "^4.19.2",