1
0
mirror of https://github.com/laurent22/joplin.git synced 2025-08-24 20:19:10 +02:00

Compare commits

...

5 Commits

Author SHA1 Message Date
Laurent Cozic
9a536d9b5e bug 2021-11-01 16:32:56 +00:00
Laurent Cozic
0929699d7d update 2021-11-01 13:18:06 +00:00
Laurent Cozic
cc9cdd0bec tests 2021-11-01 09:23:05 +00:00
Laurent Cozic
08050f6d28 sqlite 2021-11-01 08:50:37 +00:00
Laurent Cozic
63cbdd9a62 native locks 2021-11-01 08:24:51 +00:00
25 changed files with 659 additions and 141 deletions

View File

@@ -253,8 +253,12 @@ export default class JoplinServerApi {
const output = await loadResponseJson();
return output;
} catch (error) {
if (error.code !== 404) {
// Don't print error info for file not found (handled by the
// driver), or lock-acquisition errors because it's handled by
// LockHandler.
if (![404, 'hasExclusiveLock', 'hasSyncLock'].includes(error.code)) {
logger.warn(this.requestToCurl_(url, fetchOptions));
logger.warn('Code:', error.code);
logger.warn(error);
}

View File

@@ -1,5 +1,5 @@
import Logger from './Logger';
import LockHandler, { LockType } from './services/synchronizer/LockHandler';
import LockHandler, { hasActiveLock, LockType } from './services/synchronizer/LockHandler';
import Setting from './models/Setting';
import shim from './shim';
import MigrationHandler from './services/synchronizer/MigrationHandler';
@@ -298,10 +298,13 @@ export default class Synchronizer {
}
async lockErrorStatus_() {
const hasActiveExclusiveLock = await this.lockHandler().hasActiveLock(LockType.Exclusive);
const locks = await this.lockHandler().locks();
const currentDate = await this.lockHandler().currentDate();
const hasActiveExclusiveLock = await hasActiveLock(locks, currentDate, this.lockHandler().lockTtl, LockType.Exclusive);
if (hasActiveExclusiveLock) return 'hasExclusiveLock';
const hasActiveSyncLock = await this.lockHandler().hasActiveLock(LockType.Sync, this.appType_, this.clientId_);
const hasActiveSyncLock = await hasActiveLock(locks, currentDate, this.lockHandler().lockTtl, LockType.Sync, this.appType_, this.clientId_);
if (!hasActiveSyncLock) return 'syncLockGone';
return '';

View File

@@ -2,6 +2,7 @@ import { MultiPutItem } from './file-api';
import JoplinError from './JoplinError';
import JoplinServerApi from './JoplinServerApi';
import { trimSlashes } from './path-utils';
import { Lock, LockType } from './services/synchronizer/LockHandler';
// All input paths should be in the format: "path/to/file". This is converted to
// "root:/path/to/file:" when doing the API call.
@@ -40,6 +41,10 @@ export default class FileApiDriverJoplinServer {
return true;
}
public get supportsLocks() {
return true;
}
public requestRepeatCount() {
return 3;
}
@@ -196,6 +201,22 @@ export default class FileApiDriverJoplinServer {
throw new Error('Not supported');
}
public async acquireLock(type: LockType, clientType: string, clientId: string): Promise<Lock> {
return this.api().exec('POST', 'api/locks', null, {
type,
clientType: clientType,
clientId: clientId,
});
}
public async releaseLock(type: LockType, clientType: string, clientId: string) {
await this.api().exec('DELETE', `api/locks/${type}_${clientType}_${clientId}`);
}
public async listLocks() {
return this.api().exec('GET', 'api/locks');
}
public async clearRoot(path: string) {
const response = await this.list(path);
@@ -203,6 +224,8 @@ export default class FileApiDriverJoplinServer {
await this.delete(item.path);
}
await this.api().exec('POST', 'api/debug', null, { action: 'clearKeyValues' });
if (response.has_more) throw new Error('has_more support not implemented');
}
}

View File

@@ -5,6 +5,7 @@ import time from './time';
const { isHidden } = require('./path-utils');
import JoplinError from './JoplinError';
import { Lock, LockType } from './services/synchronizer/LockHandler';
const ArrayUtils = require('./ArrayUtils');
const { sprintf } = require('sprintf-js');
const Mutex = require('async-mutex').Mutex;
@@ -36,7 +37,7 @@ export interface RemoteItem {
export interface PaginatedList {
items: RemoteItem[];
has_more: boolean;
hasMore: boolean;
context: any;
}
@@ -130,6 +131,10 @@ class FileApi {
return !!this.driver().supportsAccurateTimestamp;
}
public get supportsLocks(): boolean {
return !!this.driver().supportsLocks;
}
async fetchRemoteDateOffset_() {
const tempFile = `${this.tempDirName()}/timeCheck${Math.round(Math.random() * 1000000)}.txt`;
const startTime = Date.now();
@@ -349,6 +354,22 @@ class FileApi {
logger.debug(`delta ${this.fullPath(path)}`);
return tryAndRepeat(() => this.driver_.delta(this.fullPath(path), options), this.requestRepeatCount());
}
public async acquireLock(type: LockType, clientType: string, clientId: string): Promise<Lock> {
if (!this.supportsLocks) throw new Error('Sync target does not support built-in locks');
return tryAndRepeat(() => this.driver_.acquireLock(type, clientType, clientId), this.requestRepeatCount());
}
public async releaseLock(type: LockType, clientType: string, clientId: string) {
if (!this.supportsLocks) throw new Error('Sync target does not support built-in locks');
return tryAndRepeat(() => this.driver_.releaseLock(type, clientType, clientId), this.requestRepeatCount());
}
public async listLocks() {
if (!this.supportsLocks) throw new Error('Sync target does not support built-in locks');
return tryAndRepeat(() => this.driver_.listLocks(), this.requestRepeatCount());
}
}
function basicDeltaContextFromOptions_(options: any) {

View File

@@ -3,6 +3,7 @@ import shim from '../../shim';
import JoplinError from '../../JoplinError';
import time from '../../time';
import { FileApi } from '../../file-api';
const { fileExtension, filename } = require('../../path-utils');
export enum LockType {
@@ -12,12 +13,68 @@ export enum LockType {
}
export interface Lock {
id?: string;
type: LockType;
clientType: string;
clientId: string;
updatedTime?: number;
}
function lockIsActive(lock: Lock, currentDate: Date, lockTtl: number): boolean {
return currentDate.getTime() - lock.updatedTime < lockTtl;
}
export function lockNameToObject(name: string, updatedTime: number = null): Lock {
const p = name.split('_');
return {
type: p[0] as LockType,
clientType: p[1],
clientId: p[2],
updatedTime: updatedTime,
};
}
export function hasActiveLock(locks: Lock[], currentDate: Date, lockTtl: number, lockType: LockType, clientType: string = null, clientId: string = null) {
const lock = activeLock(locks, currentDate, lockTtl, 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.
export function activeLock(locks: Lock[], currentDate: Date, lockTtl: number, lockType: LockType, clientType: string = null, clientId: string = null) {
if (lockType === LockType.Exclusive) {
const activeLocks = locks
.slice()
.filter((lock: Lock) => lockIsActive(lock, currentDate, lockTtl) && lock.type === lockType)
.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 lock = activeLocks[0];
if (clientType && clientType !== lock.clientType) return null;
if (clientId && clientId !== lock.clientId) return null;
return lock;
} else if (lockType === LockType.Sync) {
for (const lock of locks) {
if (lock.type !== lockType) continue;
if (clientType && lock.clientType !== clientType) continue;
if (clientId && lock.clientId !== clientId) continue;
if (lockIsActive(lock, currentDate, lockTtl)) return lock;
}
return null;
}
throw new Error(`Unsupported lock type: ${lockType}`);
}
export interface AcquireLockOptions {
// In theory, a client that tries to acquire an exclusive lock shouldn't
// also have a sync lock. It can however happen when the app is closed
@@ -55,14 +112,16 @@ export interface LockHandlerOptions {
lockTtl?: number;
}
export const lockDefaultTtl = 1000 * 60 * 3;
export default class LockHandler {
private api_: any = null;
private api_: FileApi = null;
private refreshTimers_: RefreshTimers = {};
private autoRefreshInterval_: number = 1000 * 60;
private lockTtl_: number = 1000 * 60 * 3;
private lockTtl_: number = lockDefaultTtl;
constructor(api: any, options: LockHandlerOptions = null) {
public constructor(api: FileApi, options: LockHandlerOptions = null) {
if (!options) options = {};
this.api_ = api;
@@ -80,6 +139,10 @@ export default class LockHandler {
this.lockTtl_ = v;
}
public get useBuiltInLocks() {
return this.api_.supportsLocks;
}
private lockFilename(lock: Lock) {
return `${[lock.type, lock.clientType, lock.clientId].join('_')}.json`;
}
@@ -97,17 +160,15 @@ export default class LockHandler {
}
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,
};
return lockNameToObject(filename(file.path), file.updated_time);
}
async locks(lockType: LockType = null): Promise<Lock[]> {
if (this.useBuiltInLocks) {
const locks = (await this.api_.listLocks()).items;
return locks;
}
const result = await this.api_.list(Dirnames.Locks);
if (result.hasMore) throw new Error('hasMore not handled'); // Shouldn't happen anyway
@@ -123,51 +184,6 @@ export default class LockHandler {
return output;
}
private lockIsActive(lock: Lock, currentDate: Date): boolean {
return currentDate.getTime() - 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);
const currentDate = await this.api_.remoteDate();
if (lockType === LockType.Exclusive) {
const activeLocks = locks
.slice()
.filter((lock: Lock) => this.lockIsActive(lock, currentDate))
.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, currentDate)) 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));
}
@@ -178,12 +194,17 @@ export default class LockHandler {
}
private async acquireSyncLock(clientType: string, clientId: string): Promise<Lock> {
if (this.useBuiltInLocks) return this.api_.acquireLock(LockType.Sync, clientType, clientId);
try {
let isFirstPass = true;
while (true) {
const locks = await this.locks();
const currentDate = await this.currentDate();
const [exclusiveLock, syncLock] = await Promise.all([
this.activeLock(LockType.Exclusive),
this.activeLock(LockType.Sync, clientType, clientId),
activeLock(locks, currentDate, this.lockTtl, LockType.Exclusive),
activeLock(locks, currentDate, this.lockTtl, LockType.Sync, clientType, clientId),
]);
if (exclusiveLock) {
@@ -222,6 +243,8 @@ export default class LockHandler {
}
private async acquireExclusiveLock(clientType: string, clientId: string, options: AcquireLockOptions = null): Promise<Lock> {
if (this.useBuiltInLocks) return this.api_.acquireLock(LockType.Exclusive, clientType, clientId);
// The logic to acquire an exclusive lock, while avoiding race conditions is as follow:
//
// - Check if there is a lock file present
@@ -252,9 +275,12 @@ export default class LockHandler {
try {
while (true) {
const locks = await this.locks();
const currentDate = await this.currentDate();
const [activeSyncLock, activeExclusiveLock] = await Promise.all([
this.activeLock(LockType.Sync),
this.activeLock(LockType.Exclusive),
activeLock(locks, currentDate, this.lockTtl, LockType.Sync),
activeLock(locks, currentDate, this.lockTtl, LockType.Exclusive),
]);
if (activeSyncLock) {
@@ -299,7 +325,11 @@ export default class LockHandler {
return [lock.type, lock.clientType, lock.clientId].join('_');
}
startAutoLockRefresh(lock: Lock, errorHandler: Function): string {
public async currentDate() {
return this.api_.remoteDate();
}
public 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}`);
@@ -321,10 +351,11 @@ export default class LockHandler {
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) {
const locks = await this.locks(lock.type);
if (!hasActiveLock(locks, await this.currentDate(), this.lockTtl, lock.type, lock.clientType, lock.clientId)) {
// If the previous lock has expired, we shouldn't try to acquire a new one. This is because other clients might have performed
// in the meantime operations that invalidates the current operation. For example, another client might have upgraded the
// sync target in the meantime, so any active operation should be cancelled here. Or if the current client was upgraded
@@ -384,6 +415,11 @@ export default class LockHandler {
}
public async releaseLock(lockType: LockType, clientType: string, clientId: string) {
if (this.useBuiltInLocks) {
await this.api_.releaseLock(lockType, clientType, clientId);
return;
}
await this.api_.delete(this.lockFilePath({
type: lockType,
clientType: clientType,

View File

@@ -1,4 +1,4 @@
import LockHandler, { LockType, LockHandlerOptions, Lock } from '../../services/synchronizer/LockHandler';
import LockHandler, { LockType, LockHandlerOptions, Lock, activeLock } from '../../services/synchronizer/LockHandler';
import { isNetworkSyncTarget, fileApi, setupDatabaseAndSynchronizer, synchronizer, switchClient, msleep, expectThrow, expectNotThrow } from '../../testing/test-utils';
// For tests with memory of file system we can use low intervals to make the tests faster.
@@ -46,6 +46,8 @@ describe('synchronizer_LockHandler', function() {
}));
it('should not use files that are not locks', (async () => {
if (lockHandler().useBuiltInLocks) return; // Doesn't make sense with built-in locks
await fileApi().put('locks/desktop.ini', 'a');
await fileApi().put('locks/exclusive.json', 'a');
await fileApi().put('locks/garbage.json', 'a');
@@ -75,10 +77,10 @@ describe('synchronizer_LockHandler', function() {
it('should auto-refresh a lock', (async () => {
const handler = newLockHandler({ autoRefreshInterval: 100 * timeoutMultipler });
const lock = await handler.acquireLock(LockType.Sync, 'desktop', '111');
const lockBefore = await handler.activeLock(LockType.Sync, 'desktop', '111');
const lockBefore = activeLock(await handler.locks(), new Date(), handler.lockTtl, LockType.Sync, 'desktop', '111');
handler.startAutoLockRefresh(lock, () => {});
await msleep(500 * timeoutMultipler);
const lockAfter = await handler.activeLock(LockType.Sync, 'desktop', '111');
const lockAfter = activeLock(await handler.locks(), new Date(), handler.lockTtl, LockType.Sync, 'desktop', '111');
expect(lockAfter.updatedTime).toBeGreaterThan(lockBefore.updatedTime);
handler.stopAutoLockRefresh(lock);
}));
@@ -95,11 +97,14 @@ describe('synchronizer_LockHandler', function() {
autoLockError = error;
});
await msleep(250 * timeoutMultipler);
try {
await msleep(250 * timeoutMultipler);
expect(autoLockError.code).toBe('lockExpired');
handler.stopAutoLockRefresh(lock);
expect(autoLockError).toBeTruthy();
expect(autoLockError.code).toBe('lockExpired');
} finally {
handler.stopAutoLockRefresh(lock);
}
}));
it('should not allow sync locks if there is an exclusive lock', (async () => {
@@ -112,6 +117,7 @@ describe('synchronizer_LockHandler', function() {
it('should not allow exclusive lock if there are sync locks', (async () => {
const lockHandler = newLockHandler({ lockTtl: 1000 * 60 * 60 });
if (lockHandler.useBuiltInLocks) return; // Tested server side
await lockHandler.acquireLock(LockType.Sync, 'mobile', '111');
await lockHandler.acquireLock(LockType.Sync, 'mobile', '222');
@@ -123,6 +129,7 @@ describe('synchronizer_LockHandler', function() {
it('should allow exclusive lock if the sync locks have expired', (async () => {
const lockHandler = newLockHandler({ lockTtl: 500 * timeoutMultipler });
if (lockHandler.useBuiltInLocks) return; // Tested server side
await lockHandler.acquireLock(LockType.Sync, 'mobile', '111');
await lockHandler.acquireLock(LockType.Sync, 'mobile', '222');
@@ -138,74 +145,44 @@ describe('synchronizer_LockHandler', function() {
const lockHandler = newLockHandler();
{
const lock1: Lock = { type: LockType.Exclusive, clientId: '1', clientType: 'd' };
const lock2: Lock = { type: LockType.Exclusive, clientId: '2', clientType: 'd' };
await lockHandler.saveLock_(lock1);
await msleep(100);
await lockHandler.saveLock_(lock2);
const locks: Lock[] = [
{
type: LockType.Exclusive,
clientId: '1',
clientType: 'd',
updatedTime: Date.now(),
},
];
const activeLock = await lockHandler.activeLock(LockType.Exclusive);
expect(activeLock.clientId).toBe('1');
await msleep(100);
locks.push({
type: LockType.Exclusive,
clientId: '2',
clientType: 'd',
updatedTime: Date.now(),
});
const lock = activeLock(locks, new Date(), lockHandler.lockTtl, LockType.Exclusive);
expect(lock.clientId).toBe('1');
}
}));
it('should ignore locks by same client when trying to acquire exclusive lock', (async () => {
const lockHandler = newLockHandler();
await lockHandler.acquireLock(LockType.Sync, 'desktop', '111');
await expectThrow(async () => {
await lockHandler.acquireLock(LockType.Exclusive, 'desktop', '111', { clearExistingSyncLocksFromTheSameClient: false });
}, 'hasSyncLock');
await expectNotThrow(async () => {
await lockHandler.acquireLock(LockType.Exclusive, 'desktop', '111', { clearExistingSyncLocksFromTheSameClient: true });
});
const activeLock = await lockHandler.activeLock(LockType.Exclusive);
expect(activeLock.clientId).toBe('111');
}));
// it('should not have race conditions', (async () => {
// it('should ignore locks by same client when trying to acquire exclusive lock', (async () => {
// const lockHandler = newLockHandler();
// const clients = [];
// for (let i = 0; i < 20; i++) {
// clients.push({
// id: 'client' + i,
// type: 'desktop',
// });
// }
// await lockHandler.acquireLock(LockType.Sync, 'desktop', '111');
// for (let loopIndex = 0; loopIndex < 1000; loopIndex++) {
// const promises:Promise<void | Lock>[] = [];
// for (let clientIndex = 0; clientIndex < clients.length; clientIndex++) {
// const client = clients[clientIndex];
// await expectThrow(async () => {
// await lockHandler.acquireLock(LockType.Exclusive, 'desktop', '111', { clearExistingSyncLocksFromTheSameClient: false });
// }, 'hasSyncLock');
// promises.push(
// lockHandler.acquireLock(LockType.Exclusive, client.type, client.id).catch(() => {})
// );
// await expectNotThrow(async () => {
// await lockHandler.acquireLock(LockType.Exclusive, 'desktop', '111', { clearExistingSyncLocksFromTheSameClient: true });
// });
// // if (gotLock) {
// // await msleep(100);
// // const locks = await lockHandler.locks(LockType.Exclusive);
// // console.info('=======================================');
// // console.info(locks);
// // lockHandler.releaseLock(LockType.Exclusive, client.type, client.id);
// // }
// // await msleep(500);
// }
// const result = await Promise.all(promises);
// const locks = result.filter((lock:any) => !!lock);
// expect(locks.length).toBe(1);
// const lock:Lock = locks[0] as Lock;
// const allLocks = await lockHandler.locks();
// console.info('================================', allLocks);
// lockHandler.releaseLock(LockType.Exclusive, lock.clientType, lock.clientId);
// }
// const lock = activeLock(await lockHandler.locks(), new Date(), lockHandler.lockTtl, LockType.Exclusive);
// expect(lock.clientId).toBe('111');
// }));
});

View File

@@ -12,7 +12,7 @@ export async function allNotesFolders() {
async function remoteItemsByTypes(types: number[]) {
const list = await fileApi().list('', { includeDirs: false, syncItemsOnly: true });
if (list.has_more) throw new Error('Not implemented!!!');
if (list.hasMore) throw new Error('Not implemented!!!');
const files = list.items;
const output = [];

View File

@@ -11,7 +11,7 @@
"devDropTables": "node dist/app.js --env dev --drop-tables",
"devDropDb": "node dist/app.js --env dev --drop-db",
"start": "node dist/app.js",
"generateTypes": "rm -f db-buildTypes.sqlite && npm run start -- --migrate-latest --env buildTypes && node dist/tools/generateTypes.js && mv db-buildTypes.sqlite schema.sqlite",
"generateTypes": "rm -f db-buildTypes.sqlite && npm run start -- --env buildTypes migrate latest && node dist/tools/generateTypes.js && mv db-buildTypes.sqlite schema.sqlite",
"tsc": "tsc --project tsconfig.json",
"test": "jest --verbose=false",
"test-ci": "npm run test",

Binary file not shown.

View File

@@ -22,6 +22,7 @@ export interface EnvVariables {
ERROR_STACK_TRACES?: string;
COOKIES_SECURE?: string;
RUNNING_IN_DOCKER?: string;
BUILTIN_LOCKS_ENABLED?: string;
// ==================================================
// URL config
@@ -211,6 +212,7 @@ export async function initConfig(envType: Env, env: EnvVariables, overrides: any
supportName: env.SUPPORT_NAME || appName,
businessEmail: env.BUSINESS_EMAIL || supportEmail,
cookieSecure: env.COOKIES_SECURE === '1',
buildInLocksEnabled: envReadBool(env.BUILTIN_LOCKS_ENABLED, false),
...overrides,
};
}

View File

@@ -116,6 +116,10 @@ export const clientType = (db: DbConnection): DatabaseConfigClient => {
return db.client.config.client;
};
export const returningSupported = (db: DbConnection) => {
return clientType(db) === DatabaseConfigClient.PostgreSQL;
};
export const isPostgres = (db: DbConnection) => {
return clientType(db) === DatabaseConfigClient.PostgreSQL;
};

View File

@@ -0,0 +1,23 @@
import { Knex } from 'knex';
import { DbConnection } from '../db';
export async function up(db: DbConnection): Promise<any> {
await db.schema.createTable('locks', (table: Knex.CreateTableBuilder) => {
table.uuid('id').unique().primary().notNullable();
table.string('user_id', 32).notNullable();
table.integer('type', 2).notNullable();
table.string('client_type', 32).notNullable();
table.string('client_id', 32).notNullable();
table.bigInteger('updated_time').notNullable();
table.bigInteger('created_time').notNullable();
});
await db.schema.alterTable('locks', (table: Knex.CreateTableBuilder) => {
table.index('user_id');
table.index('created_time');
});
}
export async function down(db: DbConnection): Promise<any> {
await db.schema.dropTable('locks');
}

View File

@@ -95,6 +95,10 @@ export default abstract class BaseModel<T> {
return this.db_;
}
protected get dbRead(): DbConnection {
return this.db;
}
protected get defaultFields(): string[] {
if (!this.defaultFields_.length) {
this.defaultFields_ = Object.keys(databaseSchema[this.tableName]);

View File

@@ -1,4 +1,6 @@
import { returningSupported } from '../db';
import { KeyValue } from '../services/database/types';
import { msleep } from '../utils/time';
import BaseModel from './BaseModel';
export enum ValueType {
@@ -6,7 +8,9 @@ export enum ValueType {
String = 2,
}
type Value = number | string;
export type Value = number | string;
export type ReadThenWriteHandler = (value: Value)=> Promise<Value>;
export default class KeyValueModel extends BaseModel<KeyValue> {
@@ -57,6 +61,45 @@ export default class KeyValueModel extends BaseModel<KeyValue> {
return this.unserializeValue(row.type, row.value) as any;
}
public async readThenWrite(key: string, handler: ReadThenWriteHandler) {
if (!returningSupported(this.db)) {
// While inside a transaction SQlite should lock the whole database
// file, which should allow atomic read then write.
await this.withTransaction(async () => {
const value: any = await this.value(key);
const newValue = await handler(value);
await this.setValue(key, newValue);
}, 'KeyValueModel::readThenWrite');
return;
}
let loopCount = 0;
while (true) {
const row: KeyValue = await this.db(this.tableName).where('key', '=', key).first();
const newValue = await handler(row ? row.value : null);
let previousValue: Value = null;
if (row) {
previousValue = row.value;
} else {
await this.setValue(key, newValue);
previousValue = newValue;
}
const updatedRows = await this
.db(this.tableName)
.update({ value: newValue }, ['id'])
.where('key', '=', key)
.where('value', '=', previousValue);
if (updatedRows.length) return;
loopCount++;
if (loopCount >= 10) throw new Error(`Could not update key: ${key}`);
await msleep(10000 * Math.random());
}
}
public async deleteValue(key: string): Promise<void> {
await this.db(this.tableName).where('key', '=', key).delete();
}
@@ -65,4 +108,8 @@ export default class KeyValueModel extends BaseModel<KeyValue> {
throw new Error('Call ::deleteValue()');
}
public async deleteAll(): Promise<void> {
await this.db(this.tableName).delete();
}
}

View File

@@ -0,0 +1,150 @@
import BaseModel, { UuidType } from './BaseModel';
import { Uuid } from '../services/database/types';
import { Lock, LockType, lockDefaultTtl, activeLock } from '@joplin/lib/services/synchronizer/LockHandler';
import { Value } from './KeyValueModel';
import { ErrorConflict } from '../utils/errors';
import uuidgen from '../utils/uuidgen';
export default class LockModel extends BaseModel<Lock> {
private lockTtl_: number = lockDefaultTtl;
protected get tableName(): string {
return 'locks';
}
protected uuidType(): UuidType {
return UuidType.Native;
}
// TODO: validate lock when acquiring and releasing
// TODO: test "should allow exclusive lock if the sync locks have expired"
// TODO: test "should not allow exclusive lock if there are sync locks"
private get lockTtl() {
return this.lockTtl_;
}
public async allLocks(userId: Uuid): Promise<Lock[]> {
const userKey = `locks::${userId}`;
const v = await this.models().keyValue().value<string>(userKey);
return v ? JSON.parse(v) : [];
}
private async acquireSyncLock(userId: Uuid, clientType: string, clientId: string): Promise<Lock> {
const userKey = `locks::${userId}`;
let output: Lock = null;
await this.models().keyValue().readThenWrite(userKey, async (value: Value) => {
let locks: Lock[] = value ? JSON.parse(value as string) : [];
const exclusiveLock = activeLock(locks, new Date(), this.lockTtl, LockType.Exclusive);
if (exclusiveLock) {
throw new ErrorConflict(`Cannot acquire lock because there is already an exclusive lock for client: ${exclusiveLock.clientType} #${exclusiveLock.clientId}`, 'hasExclusiveLock');
}
const syncLock = activeLock(locks, new Date(), this.lockTtl, LockType.Sync, clientType, clientId);
if (syncLock) {
output = {
...syncLock,
updatedTime: Date.now(),
};
locks = locks.map(l => l.id === syncLock.id ? output : l);
} else {
output = {
id: uuidgen(),
type: LockType.Sync,
clientId,
clientType,
updatedTime: Date.now(),
};
locks.push(output);
}
return JSON.stringify(locks);
});
return output;
}
private async acquireExclusiveLock(userId: Uuid, clientType: string, clientId: string): Promise<Lock> {
const userKey = `locks::${userId}`;
let output: Lock = null;
await this.models().keyValue().readThenWrite(userKey, async (value: Value) => {
let locks: Lock[] = value ? JSON.parse(value as string) : [];
const exclusiveLock = activeLock(locks, new Date(), this.lockTtl, LockType.Exclusive);
if (exclusiveLock) {
if (exclusiveLock.clientId === clientId) {
locks = locks.filter(l => l.id !== exclusiveLock.id);
output = {
...exclusiveLock,
updatedTime: Date.now(),
};
locks.push(output);
return JSON.stringify(locks);
} else {
throw new ErrorConflict(`Cannot acquire lock because there is already an exclusive lock for client: ${exclusiveLock.clientType} #${exclusiveLock.clientId}`, 'hasExclusiveLock');
}
}
const syncLock = activeLock(locks, new Date(), this.lockTtl, LockType.Sync);
if (syncLock) {
if (syncLock.clientId === clientId) {
locks = locks.filter(l => l.id !== syncLock.id);
} else {
throw new ErrorConflict(`Cannot acquire exclusive lock because there is an active sync lock for client: ${syncLock.clientType} #${syncLock.clientId}`, 'hasSyncLock');
}
}
output = {
id: uuidgen(),
type: LockType.Exclusive,
clientId,
clientType,
updatedTime: Date.now(),
};
locks.push(output);
return JSON.stringify(locks);
});
return output;
}
public async acquireLock(userId: Uuid, type: LockType, clientType: string, clientId: string): Promise<Lock> {
if (type === LockType.Sync) {
return this.acquireSyncLock(userId, clientType, clientId);
} else {
return this.acquireExclusiveLock(userId, clientType, clientId);
}
}
public async releaseLock(userId: Uuid, lockType: LockType, clientType: string, clientId: string) {
const userKey = `locks::${userId}`;
await this.models().keyValue().readThenWrite(userKey, async (value: Value) => {
const locks: Lock[] = value ? JSON.parse(value as string) : [];
for (let i = locks.length - 1; i >= 0; i--) {
const lock = locks[i];
if (lock.type === lockType && lock.clientType === clientType && lock.clientId === clientId) {
locks.splice(i, 1);
}
}
return JSON.stringify(locks);
});
}
}

View File

@@ -72,6 +72,7 @@ import SubscriptionModel from './SubscriptionModel';
import UserFlagModel from './UserFlagModel';
import EventModel from './EventModel';
import { Config } from '../utils/types';
import LockModel from './LockModel';
export class Models {
@@ -147,6 +148,10 @@ export class Models {
return new EventModel(this.db_, newModelFactory, this.config_);
}
public lock() {
return new LockModel(this.db_, newModelFactory, this.config_);
}
}
export default function newModelFactory(db: DbConnection, config: Config): Models {

View File

@@ -2,9 +2,10 @@ import config from '../../config';
import { clearDatabase, createTestUsers, CreateTestUsersOptions } from '../../tools/debugTools';
import { bodyFields } from '../../utils/requestUtils';
import Router from '../../utils/Router';
import { RouteType } from '../../utils/types';
import { Env, RouteType } from '../../utils/types';
import { SubPath } from '../../utils/routeUtils';
import { AppContext } from '../../utils/types';
import { ErrorForbidden } from '../../utils/errors';
const router = new Router(RouteType.Api);
@@ -17,7 +18,10 @@ interface Query {
}
router.post('api/debug', async (_path: SubPath, ctx: AppContext) => {
if (config().env !== Env.Dev) throw new ErrorForbidden();
const query: Query = (await bodyFields(ctx.req)) as Query;
const models = ctx.joplin.models;
console.info(`Action: ${query.action}`);
@@ -33,6 +37,10 @@ router.post('api/debug', async (_path: SubPath, ctx: AppContext) => {
if (query.action === 'clearDatabase') {
await clearDatabase(ctx.joplin.db);
}
if (query.action === 'clearKeyValues') {
await models.keyValue().deleteAll();
}
});
export default router;

View File

@@ -11,6 +11,7 @@ import { requestDeltaPagination, requestPagination } from '../../models/utils/pa
import { AclAction } from '../../models/BaseModel';
import { safeRemove } from '../../utils/fileUtils';
import { formatBytes, MB } from '../../utils/bytes';
import lockHandler from './utils/items/lockHandler';
const router = new Router(RouteType.Api);
@@ -42,6 +43,9 @@ export async function putItemContents(path: SubPath, ctx: AppContext, isBatch: b
try {
const buffer = filePath ? await fs.readFile(filePath) : Buffer.alloc(0);
const lockResult = await lockHandler(path, ctx, buffer);
if (lockResult.handled) return lockResult.response;
// This end point can optionally set the associated jop_share_id field. It
// is only useful when uploading resource blob (under .resource folder)
// since they can't have metadata. Note, Folder and Resource items all
@@ -104,8 +108,13 @@ router.del('api/items/:id', async (path: SubPath, ctx: AppContext) => {
if (ctx.joplin.env !== 'dev') throw new ErrorMethodNotAllowed('Deleting the root is not allowed');
await ctx.joplin.models.item().deleteAll(ctx.joplin.owner.id);
} else {
// const item = await itemFromPath(ctx.joplin.owner.id, ctx.joplin.models.item(), path);
// await ctx.joplin.models.item().checkIfAllowed(ctx.joplin.owner, AclAction.Delete, item);
const lockResult = await lockHandler(path, ctx);
if (lockResult.handled) return lockResult.response;
const item = await itemFromPath(ctx.joplin.owner.id, ctx.joplin.models.item(), path);
await ctx.joplin.models.item().checkIfAllowed(ctx.joplin.owner, AclAction.Delete, item);
await ctx.joplin.models.item().deleteForUser(ctx.joplin.owner.id, item);
}
} catch (error) {
@@ -137,6 +146,9 @@ router.get('api/items/:id/delta', async (_path: SubPath, ctx: AppContext) => {
});
router.get('api/items/:id/children', async (path: SubPath, ctx: AppContext) => {
const lockResult = await lockHandler(path, ctx);
if (lockResult.handled) return lockResult.response;
const itemModel = ctx.joplin.models.item();
const parentName = itemModel.pathToName(path.id);
const result = await itemModel.children(ctx.joplin.owner.id, parentName, requestPagination(ctx.query));

View File

@@ -0,0 +1,32 @@
import { lockNameToObject, LockType } from '@joplin/lib/services/synchronizer/LockHandler';
import { bodyFields } from '../../utils/requestUtils';
import Router from '../../utils/Router';
import { SubPath } from '../../utils/routeUtils';
import { AppContext, RouteType } from '../../utils/types';
const router = new Router(RouteType.Api);
interface PostFields {
type: LockType;
clientType: string;
clientId: string;
}
router.post('api/locks', async (_path: SubPath, ctx: AppContext) => {
const fields = await bodyFields<PostFields>(ctx.req);
return ctx.joplin.models.lock().acquireLock(ctx.joplin.owner.id, fields.type, fields.clientType, fields.clientId);
});
router.del('api/locks/:id', async (path: SubPath, ctx: AppContext) => {
const lock = lockNameToObject(path.id);
await ctx.joplin.models.lock().releaseLock(ctx.joplin.owner.id, lock.type, lock.clientType, lock.clientId);
});
router.get('api/locks', async (_path: SubPath, ctx: AppContext) => {
return {
items: await ctx.joplin.models.lock().allLocks(ctx.joplin.owner.id),
has_more: false,
};
});
export default router;

View File

@@ -0,0 +1,47 @@
import config from '../../../../config';
import { PaginatedItems } from '../../../../models/ItemModel';
import { Item } from '../../../../services/database/types';
import { getApi, putApi } from '../../../../utils/testing/apiUtils';
import { beforeAllDb, afterAllTests, beforeEachDb, createUserAndSession, models } from '../../../../utils/testing/testUtils';
describe('items/lockHandlers', function() {
beforeAll(async () => {
await beforeAllDb('items/lockHandlers');
});
afterAll(async () => {
await afterAllTests();
});
beforeEach(async () => {
await beforeEachDb();
config().buildInLocksEnabled = true;
});
test('should save locks to the key-value store', async function() {
const { session, user } = await createUserAndSession(1);
const lockName = 'locks/exclusive_cli_12cb74fa9de644958b2ccbc772cb4e29.json';
const now = Date.now();
const result: Item = await putApi(session.id, `items/root:/${lockName}:/content`, { testing: true });
expect(result.name).toBe(lockName);
expect(result.updated_time).toBeGreaterThanOrEqual(now);
expect(result.id).toBe(null);
const values = await models().keyValue().all();
expect(values.length).toBe(1);
expect(values[0].key).toBe(`locks::${user.id}`);
const value = JSON.parse(values[0].value);
expect(value[lockName].name).toBe(lockName);
expect(value[lockName].updated_time).toBeGreaterThanOrEqual(now);
const getResult: PaginatedItems = await getApi(session.id, 'items/root:/locks/*:children');
console.info(getResult);
expect(getResult.items[0].name).toBe(result.name);
expect(getResult.items[0].updated_time).toBe(result.updated_time);
});
});

View File

@@ -0,0 +1,98 @@
import config from '../../../../config';
import { PaginatedItems } from '../../../../models/ItemModel';
import { Value } from '../../../../models/KeyValueModel';
import { Item } from '../../../../services/database/types';
import { ErrorBadRequest } from '../../../../utils/errors';
import { SubPath } from '../../../../utils/routeUtils';
import { AppContext } from '../../../../utils/types';
interface LockHandlerResult {
handled: boolean;
response: any;
}
const lockHandler = async (path: SubPath, ctx: AppContext, requestBody: Buffer = null): Promise<LockHandlerResult | null> => {
if (!config().buildInLocksEnabled) return { handled: false, response: null };
if (!path.id || !path.id.startsWith('root:/locks/')) return { handled: false, response: null };
const ownerId = ctx.joplin.owner.id;
const models = ctx.joplin.models;
const userKey = `locks::${ownerId}`;
// PUT /api/items/root:/locks/exclusive_cli_12cb74fa9de644958b2ccbc772cb4e29.json:/content
if (ctx.method === 'PUT') {
const itemName = models.item().pathToName(path.id);
const now = Date.now();
await models.keyValue().readThenWrite(userKey, async (value: Value) => {
const output = value ? JSON.parse(value as string) : {};
output[itemName] = {
name: itemName,
updated_time: now,
jop_updated_time: now,
content: requestBody.toString(),
};
return JSON.stringify(output);
});
return {
handled: true,
response: {
[itemName]: {
item: {
name: itemName,
updated_time: now,
id: null,
},
error: null,
},
},
};
}
// DELETE /api/items/root:/locks/exclusive_cli_12cb74fa9de644958b2ccbc772cb4e29.json:
if (ctx.method === 'DELETE') {
const itemName = models.item().pathToName(path.id);
await models.keyValue().readThenWrite(userKey, async (value: Value) => {
const output = value ? JSON.parse(value as string) : {};
delete output[itemName];
return JSON.stringify(output);
});
return {
handled: true,
response: null,
};
}
// GET /api/items/root:/locks/*:/children
if (ctx.method === 'GET' && path.id === 'root:/locks/*:') {
const result = await models.keyValue().value<string>(userKey);
const obj: Record<string, Item> = result ? JSON.parse(result) : {};
const items: Item[] = [];
for (const name of Object.keys(obj)) {
items.push(obj[name]);
}
const page: PaginatedItems = {
has_more: false,
items,
};
return {
handled: true,
response: page,
};
}
throw new ErrorBadRequest(`Unhandled lock path: ${path.id}`);
};
export default lockHandler;

View File

@@ -10,6 +10,7 @@ import apiSessions from './api/sessions';
import apiShares from './api/shares';
import apiShareUsers from './api/share_users';
import apiUsers from './api/users';
import apiLocks from './api/locks';
import indexChanges from './index/changes';
import indexHelp from './index/help';
@@ -41,6 +42,7 @@ const routes: Routers = {
'api/share_users': apiShareUsers,
'api/shares': apiShares,
'api/users': apiUsers,
'api/locks': apiLocks,
'changes': indexChanges,
'home': indexHome,

View File

@@ -258,6 +258,16 @@ export interface Item extends WithDates, WithUuid {
owner_id?: Uuid;
}
export interface Lock {
id?: Uuid;
user_id?: Uuid;
type?: number;
client_type?: string;
client_id?: Uuid;
updated_time?: string;
created_time?: string;
}
export const databaseSchema: DatabaseTables = {
sessions: {
id: { type: 'string' },
@@ -430,5 +440,14 @@ export const databaseSchema: DatabaseTables = {
jop_updated_time: { type: 'string' },
owner_id: { type: 'string' },
},
locks: {
id: { type: 'string' },
user_id: { type: 'string' },
type: { type: 'number' },
client_type: { type: 'string' },
client_id: { type: 'string' },
updated_time: { type: 'string' },
created_time: { type: 'string' },
},
};
// AUTO-GENERATED-TYPES

View File

@@ -79,8 +79,8 @@ export class ErrorUnprocessableEntity extends ApiError {
export class ErrorConflict extends ApiError {
public static httpCode: number = 409;
public constructor(message: string = 'Conflict') {
super(message, ErrorConflict.httpCode);
public constructor(message: string = 'Conflict', code: string = undefined) {
super(message, ErrorConflict.httpCode, code);
Object.setPrototypeOf(this, ErrorConflict.prototype);
}
}

View File

@@ -115,6 +115,7 @@ export interface Config {
businessEmail: string;
isJoplinCloud: boolean;
cookieSecure: boolean;
buildInLocksEnabled: boolean;
}
export enum HttpMethod {