All: Add mechanism to lock and upgrade sync targets (#3524)

ReactNativeClient/lib/services/synchronizer/LockHandler.ts

@@ -0,0 +1,348 @@
+import { Dirnames } from './utils/types';
+const JoplinError = require('lib/JoplinError');
+const { time } = require('lib/time-utils');
+const { fileExtension, filename } = require('lib/path-utils.js');
+
+export enum LockType {
+	None = '',
+	Sync = 'sync',
+	Exclusive = 'exclusive',
+}
+
+export interface Lock {
+	type: LockType,
+	clientType: string,
+	clientId: string,
+	updatedTime?: number,
+}
+
+interface RefreshTimer {
+	id: any,
+	inProgress: boolean
+}
+
+interface RefreshTimers {
+	[key:string]: RefreshTimer;
+}
+
+export interface LockHandlerOptions {
+	autoRefreshInterval?: number,
+	lockTtl?: number,
+}
+
+export default class LockHandler {
+
+	private api_:any = null;
+	private refreshTimers_:RefreshTimers = {};
+	private autoRefreshInterval_:number = 1000 * 60;
+	private lockTtl_:number = 1000 * 60 * 3;
+
+	constructor(api:any, options:LockHandlerOptions = null) {
+		if (!options) options = {};
+
+		this.api_ = api;
+		if ('lockTtl' in options) this.lockTtl_ = options.lockTtl;
+		if ('autoRefreshInterval' in options) this.autoRefreshInterval_ = options.autoRefreshInterval;
+	}
+
+	public get lockTtl():number {
+		return this.lockTtl_;
+	}
+
+	// Should only be done for testing purposes since all clients should
+	// use the same lock max age.
+	public set lockTtl(v:number) {
+		this.lockTtl_ = v;
+	}
+
+	private lockFilename(lock:Lock) {
+		return `${[lock.type, lock.clientType, lock.clientId].join('_')}.json`;
+	}
+
+	private lockTypeFromFilename(name:string):LockType {
+		const ext = fileExtension(name);
+		if (ext !== 'json') return LockType.None;
+		if (name.indexOf(LockType.Sync) === 0) return LockType.Sync;
+		if (name.indexOf(LockType.Exclusive) === 0) return LockType.Exclusive;
+		return LockType.None;
+	}
+
+	private lockFilePath(lock:Lock) {
+		return `${Dirnames.Locks}/${this.lockFilename(lock)}`;
+	}
+
+	private lockFileToObject(file:any):Lock {
+		const p = filename(file.path).split('_');
+
+		return {
+			type: p[0],
+			clientType: p[1],
+			clientId: p[2],
+			updatedTime: file.updated_time,
+		};
+	}
+
+	async locks(lockType:LockType = null):Promise<Lock[]> {
+		const result = await this.api_.list(Dirnames.Locks);
+		if (result.hasMore) throw new Error('hasMore not handled'); // Shouldn't happen anyway
+
+		const output = [];
+		for (const file of result.items) {
+			const type = this.lockTypeFromFilename(file.path);
+			if (type === LockType.None) continue;
+			if (lockType && type !== lockType) continue;
+			const lock = this.lockFileToObject(file);
+			output.push(lock);
+		}
+
+		return output;
+	}
+
+	private lockIsActive(lock:Lock):boolean {
+		return Date.now() - lock.updatedTime < this.lockTtl;
+	}
+
+	async hasActiveLock(lockType:LockType, clientType:string = null, clientId:string = null) {
+		const lock = await this.activeLock(lockType, clientType, clientId);
+		return !!lock;
+	}
+
+	// Finds if there's an active lock for this clientType and clientId and returns it.
+	// If clientType and clientId are not specified, returns the first active lock
+	// of that type instead.
+	async activeLock(lockType:LockType, clientType:string = null, clientId:string = null) {
+		const locks = await this.locks(lockType);
+
+		if (lockType === LockType.Exclusive) {
+			const activeLocks = locks
+				.slice()
+				.filter((lock:Lock) => this.lockIsActive(lock))
+				.sort((a:Lock, b:Lock) => {
+					if (a.updatedTime === b.updatedTime) {
+						return a.clientId < b.clientId ? -1 : +1;
+					}
+					return a.updatedTime < b.updatedTime ? -1 : +1;
+				});
+
+			if (!activeLocks.length) return null;
+			const activeLock = activeLocks[0];
+
+			if (clientType && clientType !== activeLock.clientType) return null;
+			if (clientId && clientId !== activeLock.clientId) return null;
+			return activeLock;
+		} else if (lockType === LockType.Sync) {
+			for (const lock of locks) {
+				if (clientType && lock.clientType !== clientType) continue;
+				if (clientId && lock.clientId !== clientId) continue;
+				if (this.lockIsActive(lock)) return lock;
+			}
+			return null;
+		}
+
+		throw new Error(`Unsupported lock type: ${lockType}`);
+	}
+
+	private async saveLock(lock:Lock) {
+		await this.api_.put(this.lockFilePath(lock), JSON.stringify(lock));
+	}
+
+	// This is for testing only
+	public async saveLock_(lock:Lock) {
+		return this.saveLock(lock);
+	}
+
+	private async acquireSyncLock(clientType:string, clientId:string):Promise<Lock> {
+		try {
+			let isFirstPass = true;
+			while (true) {
+				const [exclusiveLock, syncLock] = await Promise.all([
+					this.activeLock(LockType.Exclusive),
+					this.activeLock(LockType.Sync, clientType, clientId),
+				]);
+
+				if (exclusiveLock) {
+					throw new JoplinError(`Cannot acquire sync lock because the following client has an exclusive lock on the sync target: ${this.lockToClientString(exclusiveLock)}`, 'hasExclusiveLock');
+				}
+
+				if (syncLock) {
+					// Normally the second pass should happen immediately afterwards, but if for some reason
+					// (slow network, etc.) it took more than 10 seconds then refresh the lock.
+					if (isFirstPass || Date.now() - syncLock.updatedTime > 1000 * 10) {
+						await this.saveLock(syncLock);
+					}
+					return syncLock;
+				}
+
+				// Something wrong happened, which means we saved a lock but we didn't read
+				// it back. Could be application error or server issue.
+				if (!isFirstPass) throw new Error('Cannot acquire sync lock: either the lock could be written but not read back. Or it was expired before it was read again.');
+
+				await this.saveLock({
+					type: LockType.Sync,
+					clientType: clientType,
+					clientId: clientId,
+				});
+
+				isFirstPass = false;
+			}
+		} catch (error) {
+			await this.releaseLock(LockType.Sync, clientType, clientId);
+			throw error;
+		}
+	}
+
+	private lockToClientString(lock:Lock):string {
+		return `(${lock.clientType} #${lock.clientId})`;
+	}
+
+	private async acquireExclusiveLock(clientType:string, clientId:string, timeoutMs:number = 0):Promise<Lock> {
+		// The logic to acquire an exclusive lock, while avoiding race conditions is as follow:
+		//
+		// - Check if there is a lock file present
+		//
+		// - If there is a lock file, see if I'm the one owning it by checking that its content has my identifier.
+		// - If that's the case, just write to the data file then delete the lock file.
+		// - If that's not the case, just wait a second or a small random length of time and try the whole cycle again-.
+		//
+		// -If there is no lock file, create one with my identifier and try the whole cycle again to avoid race condition (re-check that the lock file is really mine)-.
+
+		const startTime = Date.now();
+
+		async function waitForTimeout() {
+			if (!timeoutMs) return false;
+
+			const elapsed = Date.now() - startTime;
+			if (timeoutMs && elapsed < timeoutMs) {
+				await time.sleep(2);
+				return true;
+			}
+			return false;
+		}
+
+		try {
+			while (true) {
+				const [activeSyncLock, activeExclusiveLock] = await Promise.all([
+					this.activeLock(LockType.Sync),
+					this.activeLock(LockType.Exclusive),
+				]);
+
+				if (activeSyncLock) {
+					if (await waitForTimeout()) continue;
+					throw new JoplinError(`Cannot acquire exclusive lock because the following clients have a sync lock on the target: ${this.lockToClientString(activeSyncLock)}`, 'hasSyncLock');
+				}
+
+				if (activeExclusiveLock) {
+					if (activeExclusiveLock.clientId === clientId) {
+						// Save it again to refresh the timestamp
+						await this.saveLock(activeExclusiveLock);
+						return activeExclusiveLock;
+					} else {
+						// If there's already an exclusive lock, wait for it to be released
+						if (await waitForTimeout()) continue;
+						throw new JoplinError(`Cannot acquire exclusive lock because the following client has an exclusive lock on the sync target: ${this.lockToClientString(activeExclusiveLock)}`, 'hasExclusiveLock');
+					}
+				} else {
+					// If there's not already an exclusive lock, acquire one
+					// then loop again to check that we really got the lock
+					// (to prevent race conditions)
+					await this.saveLock({
+						type: LockType.Exclusive,
+						clientType: clientType,
+						clientId: clientId,
+					});
+
+					await time.msleep(100);
+				}
+			}
+		} catch (error) {
+			await this.releaseLock(LockType.Exclusive, clientType, clientId);
+			throw error;
+		}
+	}
+
+	private autoLockRefreshHandle(lock:Lock) {
+		return [lock.type, lock.clientType, lock.clientId].join('_');
+	}
+
+	startAutoLockRefresh(lock:Lock, errorHandler:Function):string {
+		const handle = this.autoLockRefreshHandle(lock);
+		if (this.refreshTimers_[handle]) {
+			throw new Error(`There is already a timer refreshing this lock: ${handle}`);
+		}
+
+		this.refreshTimers_[handle] = {
+			id: null,
+			inProgress: false,
+		};
+
+		this.refreshTimers_[handle].id = setInterval(async () => {
+			if (this.refreshTimers_[handle].inProgress) return;
+
+			const defer = () => {
+				if (!this.refreshTimers_[handle]) return;
+				this.refreshTimers_[handle].inProgress = false;
+			};
+
+			this.refreshTimers_[handle].inProgress = true;
+
+			let error = null;
+			const hasActiveLock = await this.hasActiveLock(lock.type, lock.clientType, lock.clientId);
+			if (!this.refreshTimers_[handle]) return defer(); // Timeout has been cleared
+
+			if (!hasActiveLock) {
+				error = new JoplinError('Lock has expired', 'lockExpired');
+			} else {
+				try {
+					await this.acquireLock(lock.type, lock.clientType, lock.clientId);
+					if (!this.refreshTimers_[handle]) return defer(); // Timeout has been cleared
+				} catch (e) {
+					error = e;
+				}
+			}
+
+			if (error) {
+				if (this.refreshTimers_[handle]) {
+					clearInterval(this.refreshTimers_[handle].id);
+					delete this.refreshTimers_[handle];
+				}
+				errorHandler(error);
+			}
+
+			defer();
+		}, this.autoRefreshInterval_);
+
+		return handle;
+	}
+
+	stopAutoLockRefresh(lock:Lock) {
+		const handle = this.autoLockRefreshHandle(lock);
+		if (!this.refreshTimers_[handle]) {
+			// Should not throw an error because lock may have been cleared in startAutoLockRefresh
+			// if there was an error.
+			// throw new Error(`There is no such lock being auto-refreshed: ${this.lockToString(lock)}`);
+			return;
+		}
+
+		clearInterval(this.refreshTimers_[handle].id);
+		delete this.refreshTimers_[handle];
+	}
+
+	async acquireLock(lockType:LockType, clientType:string, clientId:string, timeoutMs:number = 0):Promise<Lock> {
+		if (lockType === LockType.Sync) {
+			return this.acquireSyncLock(clientType, clientId);
+		} else if (lockType === LockType.Exclusive) {
+			return this.acquireExclusiveLock(clientType, clientId, timeoutMs);
+		} else {
+			throw new Error(`Invalid lock type: ${lockType}`);
+		}
+	}
+
+	async releaseLock(lockType:LockType, clientType:string, clientId:string) {
+		await this.api_.delete(this.lockFilePath({
+			type: lockType,
+			clientType: clientType,
+			clientId: clientId,
+		}));
+	}
+
+}

ReactNativeClient/lib/services/synchronizer/MigrationHandler.ts

@@ -0,0 +1,141 @@
+import LockHandler, { LockType } from './LockHandler';
+import { Dirnames } from './utils/types';
+const BaseService = require('lib/services/BaseService.js');
+
+// To add a new migration:
+// - Add the migration logic in ./migrations/VERSION_NUM.js
+// - Add the file to the array below.
+// - Set Setting.syncVersion to VERSION_NUM in models/Setting.js
+// - Add tests in synchronizer_migrationHandler
+const migrations = [
+	null,
+	require('./migrations/1.js').default,
+	require('./migrations/2.js').default,
+];
+
+const Setting = require('lib/models/Setting');
+const { sprintf } = require('sprintf-js');
+const JoplinError = require('lib/JoplinError');
+
+interface SyncTargetInfo {
+	version: number,
+}
+
+export default class MigrationHandler extends BaseService {
+
+	private api_:any = null;
+	private lockHandler_:LockHandler = null;
+	private clientType_:string;
+	private clientId_:string;
+
+	constructor(api:any, lockHandler:LockHandler, clientType:string, clientId:string) {
+		super();
+		this.api_ = api;
+		this.lockHandler_ = lockHandler;
+		this.clientType_ = clientType;
+		this.clientId_ = clientId;
+	}
+
+	public async fetchSyncTargetInfo():Promise<SyncTargetInfo> {
+		const syncTargetInfoText = await this.api_.get('info.json');
+
+		// Returns version 0 if the sync target is empty
+		let output:SyncTargetInfo = { version: 0 };
+
+		if (syncTargetInfoText) {
+			output = JSON.parse(syncTargetInfoText);
+			if (!output.version) throw new Error('Missing "version" field in info.json');
+		} else {
+			const oldVersion = await this.api_.get('.sync/version.txt');
+			if (oldVersion) output = { version: 1 };
+		}
+
+		return output;
+	}
+
+	private serializeSyncTargetInfo(info:SyncTargetInfo) {
+		return JSON.stringify(info);
+	}
+
+	async checkCanSync():Promise<SyncTargetInfo> {
+		const supportedSyncTargetVersion = Setting.value('syncVersion');
+		const syncTargetInfo = await this.fetchSyncTargetInfo();
+
+		if (syncTargetInfo.version) {
+			if (syncTargetInfo.version > supportedSyncTargetVersion) {
+				throw new JoplinError(sprintf('Sync version of the target (%d) is greater than the version supported by the client (%d). Please upgrade your client.', syncTargetInfo.version, supportedSyncTargetVersion), 'outdatedClient');
+			} else if (syncTargetInfo.version < supportedSyncTargetVersion) {
+				throw new JoplinError(sprintf('Sync version of the target (%d) is lower than the version supported by the client (%d). Please upgrade the sync target.', syncTargetInfo.version, supportedSyncTargetVersion), 'outdatedSyncTarget');
+			}
+		}
+
+		return syncTargetInfo;
+	}
+
+	async upgrade(targetVersion:number = 0) {
+		const supportedSyncTargetVersion = Setting.value('syncVersion');
+		const syncTargetInfo = await this.fetchSyncTargetInfo();
+
+		if (syncTargetInfo.version > supportedSyncTargetVersion) {
+			throw new JoplinError(sprintf('Sync version of the target (%d) is greater than the version supported by the client (%d). Please upgrade your client.', syncTargetInfo.version, supportedSyncTargetVersion), 'outdatedClient');
+		}
+
+		// if (supportedSyncTargetVersion !== migrations.length - 1) {
+		// 	// Sanity check - it means a migration has been added by syncVersion has not be incremented or vice-versa,
+		// 	// so abort as it can cause strange issues.
+		// 	throw new JoplinError('Application error: mismatch between max supported sync version and max migration number: ' + supportedSyncTargetVersion + ' / ' + (migrations.length - 1));
+		// }
+
+		// Special case for version 1 because it didn't have the lock folder and without
+		// it the lock handler will break. So we create the directory now.
+		// Also if the sync target version is 0, it means it's a new one so we need the
+		// lock folder first before doing anything else.
+		if (syncTargetInfo.version === 0 || syncTargetInfo.version === 1) {
+			this.logger().info('MigrationHandler: Sync target version is 0 or 1 - creating "locks" directory:', syncTargetInfo);
+			await this.api_.mkdir(Dirnames.Locks);
+		}
+
+		this.logger().info('MigrationHandler: Acquiring exclusive lock');
+		const exclusiveLock = await this.lockHandler_.acquireLock(LockType.Exclusive, this.clientType_, this.clientId_, 1000 * 30);
+		let autoLockError = null;
+		this.lockHandler_.startAutoLockRefresh(exclusiveLock, (error:any) => {
+			autoLockError = error;
+		});
+
+		this.logger().info('MigrationHandler: Acquired exclusive lock:', exclusiveLock);
+
+		try {
+			for (let newVersion = syncTargetInfo.version + 1; newVersion < migrations.length; newVersion++) {
+				if (targetVersion && newVersion > targetVersion) break;
+
+				const fromVersion = newVersion - 1;
+
+				this.logger().info(`MigrationHandler: Migrating from version ${fromVersion} to version ${newVersion}`);
+
+				const migration = migrations[newVersion];
+				if (!migration) continue;
+
+				try {
+					if (autoLockError) throw autoLockError;
+					await migration(this.api_);
+					if (autoLockError) throw autoLockError;
+
+					await this.api_.put('info.json', this.serializeSyncTargetInfo({
+						...syncTargetInfo,
+						version: newVersion,
+					}));
+
+					this.logger().info(`MigrationHandler: Done migrating from version ${fromVersion} to version ${newVersion}`);
+				} catch (error) {
+					error.message = `Could not upgrade from version ${fromVersion} to version ${newVersion}: ${error.message}`;
+					throw error;
+				}
+			}
+		} finally {
+			this.logger().info('MigrationHandler: Releasing exclusive lock');
+			this.lockHandler_.stopAutoLockRefresh(exclusiveLock);
+			await this.lockHandler_.releaseLock(LockType.Exclusive, this.clientType_, this.clientId_);
+		}
+	}
+
+}

ReactNativeClient/lib/services/synchronizer/gui/useSyncTargetUpgrade.ts

@@ -0,0 +1,50 @@
+import { useEffect, useState } from 'react';
+import MigrationHandler from 'lib/services/synchronizer/MigrationHandler';
+const Setting = require('lib/models/Setting');
+const { reg } = require('lib/registry');
+
+export interface SyncTargetUpgradeResult {
+	done: boolean,
+	error: any,
+}
+
+export default function useSyncTargetUpgrade():SyncTargetUpgradeResult {
+	const [upgradeResult, setUpgradeResult] = useState<SyncTargetUpgradeResult>({
+		done: false,
+		error: null,
+	});
+
+	async function upgradeSyncTarget() {
+		let error = null;
+		try {
+			const synchronizer = await reg.syncTarget().synchronizer();
+
+			const migrationHandler = new MigrationHandler(
+				synchronizer.api(),
+				synchronizer.lockHandler(),
+				Setting.value('appType'),
+				Setting.value('clientId')
+			);
+
+			await migrationHandler.upgrade();
+		} catch (e) {
+			error = e;
+		}
+
+		if (!error) {
+			Setting.setValue('sync.upgradeState', Setting.SYNC_UPGRADE_STATE_IDLE);
+			await Setting.saveAll();
+		}
+
+		setUpgradeResult({
+			done: true,
+			error: error,
+		});
+	}
+
+	useEffect(function() {
+		upgradeSyncTarget();
+	}, []);
+
+	return upgradeResult;
+}

ReactNativeClient/lib/services/synchronizer/migrations/1.ts

@@ -0,0 +1,9 @@
+export default async function(api:any) {
+	await Promise.all([
+		api.mkdir('.resource'),
+		api.mkdir('.sync'),
+		api.mkdir('.lock'),
+	]);
+
+	await api.put('.sync/version.txt', '1');
+}

ReactNativeClient/lib/services/synchronizer/migrations/2.ts

@@ -0,0 +1,10 @@
+import { Dirnames } from '../utils/types';
+
+export default async function(api:any) {
+	await Promise.all([
+		api.put('.sync/version.txt', '2'),
+		api.put('.sync/readme.txt', '2020-07-16: In the new sync format, the version number is stored in /info.json. However, for backward compatibility, we need to keep the old version.txt file here, otherwise old clients will automatically recreate it, and assume a sync target version 1. So we keep it here but set its value to "2", so that old clients know that they need to be upgraded. This directory can be removed after a year or so, once we are confident that all clients have been upgraded to recent versions.'),
+		api.mkdir(Dirnames.Locks),
+		api.mkdir(Dirnames.Temp),
+	]);
+}

ReactNativeClient/lib/services/synchronizer/utils/types.ts

@@ -0,0 +1,6 @@
+// eslint-disable-next-line import/prefer-default-export
+export enum Dirnames {
+	Locks = 'locks',
+	Resources = '.resource',
+	Temp = 'temp',
+}