Modified:

- Plugins and settings is now in beta.

Implemented:
- Show the count of the pending processes into the status.

README.md

@@ -75,6 +75,8 @@ Synchronization status is shown in statusbar.
     -   ⚠ Error occurred.
 -   ↑ Uploaded pieces
 -   ↓ Downloaded pieces
+-   ⏳ Count of the pending process  
+If you have deleted or renamed files, please wait until this disappears.
 
 # More supplements
 

README_ja.md

@@ -78,6 +78,8 @@ Self-hosted LiveSync用にWebClipperも作りました。Chrome Web Storeから
   - ⚠ エラーが発生しています
 - ↑ 送信したデータ数
 - ↓ 受信したデータ数
+-   ⏳ 保留している処理の数です  
+ファイルを削除したりリネームした場合、この表示が消えるまでお待ちください。
 
 # さらなる補足
 - ファイルは同期された後、タイムスタンプを比較して新しければいったん新しい方で上書きされます。その後、衝突が発生したかによって、マージが行われます。

manifest.json

@@ -1,7 +1,7 @@
 {
     "id": "obsidian-livesync",
     "name": "Self-hosted LiveSync",
-    "version": "0.4.0",
+    "version": "0.5.0",
     "minAppVersion": "0.9.12",
     "description": "Community implementation of self-hosted livesync. Reflect your vault changes to some other devices immediately. Please make sure to disable other synchronize solutions to avoid content corruption or duplication.",
     "author": "vorotamoroz",

package-lock.json

@@ -1,12 +1,12 @@
 {
     "name": "obsidian-livesync",
-    "version": "0.4.0",
+    "version": "0.5.0",
     "lockfileVersion": 2,
     "requires": true,
     "packages": {
         "": {
             "name": "obsidian-livesync",
-            "version": "0.4.0",
+            "version": "0.5.0",
             "license": "MIT",
             "dependencies": {
                 "diff-match-patch": "^1.0.5",

package.json

@@ -1,6 +1,6 @@
 {
     "name": "obsidian-livesync",
-    "version": "0.4.0",
+    "version": "0.5.0",
     "description": "Reflect your vault changes to some other devices immediately. Please make sure to disable other synchronize solutions to avoid content corruption or duplication.",
     "main": "main.js",
     "scripts": {

src/LocalPouchDB.ts

@@ -25,7 +25,7 @@ import {
 } from "./types";
 import { resolveWithIgnoreKnownError, delay, path2id, runWithLock } from "./utils";
 import { Logger } from "./logger";
-import { checkRemoteVersion, connectRemoteCouchDB } from "./utils_couchdb";
+import { checkRemoteVersion, connectRemoteCouchDB, getLastPostFailedBySize } from "./utils_couchdb";
 import { decrypt, encrypt } from "./e2ee";
 
 export class LocalPouchDB {
@@ -763,6 +763,12 @@ export class LocalPouchDB {
             }
 
             const syncOptionBase: PouchDB.Replication.SyncOptions = {
+                pull: {
+                    checkpoint: "target",
+                },
+                push: {
+                    checkpoint: "source",
+                },
                 batches_limit: setting.batches_limit,
                 batch_size: setting.batch_size,
             };
@@ -770,7 +776,7 @@ export class LocalPouchDB {
             const db = dbret.db;
             const totalCount = (await this.localDatabase.info()).doc_count;
             //replicate once
-            const replicate = this.localDatabase.replicate.to(db, syncOptionBase);
+            const replicate = this.localDatabase.replicate.to(db, { checkpoint: "source", ...syncOptionBase });
             replicate
                 .on("active", () => {
                     this.syncStatus = "CONNECTED";
@@ -806,7 +812,7 @@ export class LocalPouchDB {
         });
     }
 
-    async checkReplicationConnectivity(setting: ObsidianLiveSyncSettings, keepAlive: boolean) {
+    async checkReplicationConnectivity(setting: ObsidianLiveSyncSettings, keepAlive: boolean, skipCheck: boolean) {
         if (!this.isReady) {
             Logger("Database is not ready.");
             return false;
@@ -832,40 +838,41 @@ export class LocalPouchDB {
             return false;
         }
 
-        if (!(await checkRemoteVersion(dbret.db, this.migrate.bind(this), VER))) {
+        if (!skipCheck) {
-            Logger("Remote database is newer or corrupted, make sure to latest version of self-hosted-livesync installed", LOG_LEVEL.NOTICE);
+            if (!(await checkRemoteVersion(dbret.db, this.migrate.bind(this), VER))) {
-            return false;
+                Logger("Remote database is newer or corrupted, make sure to latest version of self-hosted-livesync installed", LOG_LEVEL.NOTICE);
+                return false;
+            }
+
+            const defMilestonePoint: EntryMilestoneInfo = {
+                _id: MILSTONE_DOCID,
+                type: "milestoneinfo",
+                created: (new Date() as any) / 1,
+                locked: false,
+                accepted_nodes: [this.nodeid],
+            };
+            // const remoteInfo = dbret.info;
+            // const localInfo = await this.localDatabase.info();
+            // const remoteDocsCount = remoteInfo.doc_count;
+            // const localDocsCount = localInfo.doc_count;
+            // const remoteUpdSeq = typeof remoteInfo.update_seq == "string" ? Number(remoteInfo.update_seq.split("-")[0]) : remoteInfo.update_seq;
+            // const localUpdSeq = typeof localInfo.update_seq == "string" ? Number(localInfo.update_seq.split("-")[0]) : localInfo.update_seq;
+
+            // Logger(`Database diffences: remote:${remoteDocsCount} docs / last update ${remoteUpdSeq}`);
+            // Logger(`Database diffences: local :${localDocsCount} docs / last update ${localUpdSeq}`);
+
+            const remoteMilestone: EntryMilestoneInfo = await resolveWithIgnoreKnownError(dbret.db.get(MILSTONE_DOCID), defMilestonePoint);
+            this.remoteLocked = remoteMilestone.locked;
+            this.remoteLockedAndDeviceNotAccepted = remoteMilestone.locked && remoteMilestone.accepted_nodes.indexOf(this.nodeid) == -1;
+
+            if (remoteMilestone.locked && remoteMilestone.accepted_nodes.indexOf(this.nodeid) == -1) {
+                Logger("Remote database marked as 'Auto Sync Locked'. And this devide does not marked as resolved device. see settings dialog.", LOG_LEVEL.NOTICE);
+                return false;
+            }
+            if (typeof remoteMilestone._rev == "undefined") {
+                await dbret.db.put(remoteMilestone);
+            }
         }
-
-        const defMilestonePoint: EntryMilestoneInfo = {
-            _id: MILSTONE_DOCID,
-            type: "milestoneinfo",
-            created: (new Date() as any) / 1,
-            locked: false,
-            accepted_nodes: [this.nodeid],
-        };
-        // const remoteInfo = dbret.info;
-        // const localInfo = await this.localDatabase.info();
-        // const remoteDocsCount = remoteInfo.doc_count;
-        // const localDocsCount = localInfo.doc_count;
-        // const remoteUpdSeq = typeof remoteInfo.update_seq == "string" ? Number(remoteInfo.update_seq.split("-")[0]) : remoteInfo.update_seq;
-        // const localUpdSeq = typeof localInfo.update_seq == "string" ? Number(localInfo.update_seq.split("-")[0]) : localInfo.update_seq;
-
-        // Logger(`Database diffences: remote:${remoteDocsCount} docs / last update ${remoteUpdSeq}`);
-        // Logger(`Database diffences: local :${localDocsCount} docs / last update ${localUpdSeq}`);
-
-        const remoteMilestone: EntryMilestoneInfo = await resolveWithIgnoreKnownError(dbret.db.get(MILSTONE_DOCID), defMilestonePoint);
-        this.remoteLocked = remoteMilestone.locked;
-        this.remoteLockedAndDeviceNotAccepted = remoteMilestone.locked && remoteMilestone.accepted_nodes.indexOf(this.nodeid) == -1;
-
-        if (remoteMilestone.locked && remoteMilestone.accepted_nodes.indexOf(this.nodeid) == -1) {
-            Logger("Remote database marked as 'Auto Sync Locked'. And this devide does not marked as resolved device. see settings dialog.", LOG_LEVEL.NOTICE);
-            return false;
-        }
-        if (typeof remoteMilestone._rev == "undefined") {
-            await dbret.db.put(remoteMilestone);
-        }
-
         const syncOptionBase: PouchDB.Replication.SyncOptions = {
             batches_limit: setting.batches_limit,
             batch_size: setting.batch_size,
@@ -877,33 +884,49 @@ export class LocalPouchDB {
 
     async openReplication(setting: ObsidianLiveSyncSettings, keepAlive: boolean, showResult: boolean, callback: (e: PouchDB.Core.ExistingDocument<EntryDoc>[]) => Promise<void>): Promise<boolean> {
         return await runWithLock("replicate", false, () => {
-            return this._openReplication(setting, keepAlive, showResult, callback);
+            return this._openReplication(setting, keepAlive, showResult, callback, false);
         });
     }
 
-    async _openReplication(setting: ObsidianLiveSyncSettings, keepAlive: boolean, showResult: boolean, callback: (e: PouchDB.Core.ExistingDocument<EntryDoc>[]) => Promise<void>): Promise<boolean> {
+    originalSetting: ObsidianLiveSyncSettings = null;
-        const ret = await this.checkReplicationConnectivity(setting, keepAlive);
+    // last_seq: number = 200;
+    async _openReplication(setting: ObsidianLiveSyncSettings, keepAlive: boolean, showResult: boolean, callback: (e: PouchDB.Core.ExistingDocument<EntryDoc>[]) => Promise<void>, retrying: boolean): Promise<boolean> {
+        const ret = await this.checkReplicationConnectivity(setting, keepAlive, retrying);
         if (ret === false) return false;
         let notice: Notice = null;
         if (showResult) {
-            notice = new Notice("Replicating", 0);
+            notice = new Notice("Looking for the point last synchronized point.", 0);
         }
         const { db, syncOptionBase, syncOption } = ret;
         //replicate once
         this.syncStatus = "STARTED";
+        this.updateInfo();
 
         let resolved = false;
         const docArrivedOnStart = this.docArrived;
         const docSentOnStart = this.docSent;
+
         const _openReplicationSync = () => {
             Logger("Sync Main Started");
+            if (!retrying) {
+                this.originalSetting = setting;
+            }
             this.syncHandler = this.cancelHandler(this.syncHandler);
-            this.syncHandler = this.localDatabase.sync<EntryDoc>(db, syncOption);
+            this.syncHandler = this.localDatabase.sync<EntryDoc>(db, {
+                ...syncOption,
+                pull: {
+                    checkpoint: "target",
+                },
+                push: {
+                    checkpoint: "source",
+                },
+            });
             this.syncHandler
                 .on("active", () => {
                     this.syncStatus = "CONNECTED";
                     this.updateInfo();
                     Logger("Replication activated");
+                    if (notice != null) notice.setMessage(`Activated..`);
                 })
                 .on("change", async (e) => {
                     try {
@@ -924,6 +947,16 @@ export class LocalPouchDB {
                         Logger("Replication callback error");
                         Logger(ex);
                     }
+                    // re-connect to retry with original setting
+                    if (retrying) {
+                        if (this.docSent - docSentOnStart + (this.docArrived - docArrivedOnStart) > this.originalSetting.batch_size * 2) {
+                            // restore sync values
+                            Logger("Back into original settings once.");
+                            if (notice != null) notice.hide();
+                            this.syncHandler = this.cancelHandler(this.syncHandler);
+                            this._openReplication(this.originalSetting, keepAlive, showResult, callback, false);
+                        }
+                    }
                 })
                 .on("complete", (e) => {
                     this.syncStatus = "COMPLETED";
@@ -948,8 +981,25 @@ export class LocalPouchDB {
                     this.syncHandler = this.cancelHandler(this.syncHandler);
                     this.updateInfo();
                     if (notice != null) notice.hide();
-                    Logger("Replication error", LOG_LEVEL.NOTICE);
+                    if (getLastPostFailedBySize()) {
-                    Logger(e);
+                        if (keepAlive) {
+                            Logger("Replication stopped.", LOG_LEVEL.NOTICE);
+                        } else {
+                            // Duplicate settings for smaller batch.
+                            const xsetting: ObsidianLiveSyncSettings = JSON.parse(JSON.stringify(setting));
+                            xsetting.batch_size = Math.ceil(xsetting.batch_size / 2);
+                            xsetting.batches_limit = Math.ceil(xsetting.batches_limit / 2);
+                            if (xsetting.batch_size <= 3 || xsetting.batches_limit <= 3) {
+                                Logger("We can't replicate more lower value.", showResult ? LOG_LEVEL.NOTICE : LOG_LEVEL.INFO);
+                            } else {
+                                Logger(`Retry with lower batch size:${xsetting.batch_size}/${xsetting.batches_limit}`, showResult ? LOG_LEVEL.NOTICE : LOG_LEVEL.INFO);
+                                this._openReplication(xsetting, keepAlive, showResult, callback, true);
+                            }
+                        }
+                    } else {
+                        Logger("Replication error", LOG_LEVEL.NOTICE);
+                        Logger(e);
+                    }
                 })
                 .on("paused", (e) => {
                     this.syncStatus = "PAUSED";
@@ -974,7 +1024,7 @@ export class LocalPouchDB {
         Logger(await db.info(), LOG_LEVEL.VERBOSE);
         let replicate: PouchDB.Replication.Replication<EntryDoc>;
         try {
-            replicate = this.localDatabase.replicate.from(db, syncOptionBase);
+            replicate = this.localDatabase.replicate.from(db, { checkpoint: "target", ...syncOptionBase });
             replicate
                 .on("active", () => {
                     this.syncStatus = "CONNECTED";

src/ObsidianLiveSyncSettingTab.ts

@@ -811,7 +811,7 @@ export class ObsidianLiveSyncSettingTab extends PluginSettingTab {
         // With great respect, thank you TfTHacker!
         // refered: https://github.com/TfTHacker/obsidian42-brat/blob/main/src/features/BetaPlugins.ts
         const containerPluginSettings = containerEl.createDiv();
-        containerPluginSettings.createEl("h3", { text: "Plugins and settings (bleeding edge)" });
+        containerPluginSettings.createEl("h3", { text: "Plugins and settings (beta)" });
 
         const updateDisabledOfDeviceAndVaultName = () => {
             vaultName.setDisabled(this.plugin.settings.autoSweepPlugins || this.plugin.settings.autoSweepPluginsPeriodic);

src/main.ts

@@ -18,7 +18,7 @@ import {
     diff_result,
     FLAGMD_REDFLAG,
 } from "./types";
-import { base64ToString, arrayBufferToBase64, base64ToArrayBuffer, isValidPath, versionNumberString2Number, id2path, path2id, runWithLock, shouldBeIgnored } from "./utils";
+import { base64ToString, arrayBufferToBase64, base64ToArrayBuffer, isValidPath, versionNumberString2Number, id2path, path2id, runWithLock, shouldBeIgnored, getProcessingCounts, setLockNotifier } from "./utils";
 import { Logger, setLogger } from "./logger";
 import { LocalPouchDB } from "./LocalPouchDB";
 import { LogDisplayModal } from "./LogDisplayModal";
@@ -205,6 +205,9 @@ export default class ObsidianLiveSyncPlugin extends Plugin {
         });
         this.triggerRealizeSettingSyncMode = debounce(this.triggerRealizeSettingSyncMode.bind(this), 1000);
         this.triggerCheckPluginUpdate = debounce(this.triggerCheckPluginUpdate.bind(this), 3000);
+        setLockNotifier(() => {
+            this.refreshStatusText();
+        });
     }
     onunload() {
         this.localDatabase.onunload();
@@ -787,7 +790,9 @@ export default class ObsidianLiveSyncPlugin extends Plugin {
             waiting = " " + this.batchFileChange.map((e) => "🛫").join("");
             waiting = waiting.replace(/(🛫){10}/g, "🚀");
         }
-        const message = `Sync:${w} ↑${sent} ↓${arrived}${waiting}`;
+        const procs = getProcessingCounts();
+        const procsDisp = procs==0?"":` ⏳${procs}`;
+        const message = `Sync:${w} ↑${sent} ↓${arrived}${waiting}${procsDisp}`;
         this.setStatusBarText(message);
     }
     setStatusBarText(message: string) {

src/utils.ts

@@ -121,7 +121,28 @@ function objectToKey(key: any): string {
     const keys = Object.keys(key).sort((a, b) => a.localeCompare(b));
     return keys.map((e) => e + objectToKey(key[e])).join(":");
 }
+export function getProcessingCounts() {
+    let count = 0;
+    for (const v in pendingProcs) {
+        count += pendingProcs[v].length;
+    }
+    count += runningProcs.length;
+    return count;
+}
 
+let externalNotifier: () => void = () => {};
+let notifyTimer: number = null;
+export function setLockNotifier(fn: () => void) {
+    externalNotifier = fn;
+}
+function notifyLock() {
+    if (notifyTimer != null) {
+        window.clearTimeout(notifyTimer);
+    }
+    notifyTimer = window.setTimeout(() => {
+        externalNotifier();
+    }, 100);
+}
 // Just run async/await as like transacion ISOLATION SERIALIZABLE
 export function runWithLock<T>(key: unknown, ignoreWhenRunning: boolean, proc: () => Promise<T>): Promise<T> {
     // Logger(`Lock:${key}:enter`, LOG_LEVEL.VERBOSE);
@@ -130,11 +151,13 @@ export function runWithLock<T>(key: unknown, ignoreWhenRunning: boolean, proc: (
         if (typeof pendingProcs[lockKey] === "undefined") {
             //simply unlock
             runningProcs.remove(lockKey);
+            notifyLock();
             // Logger(`Lock:${lockKey}:released`, LOG_LEVEL.VERBOSE);
         } else {
             Logger(`Lock:${lockKey}:left ${pendingProcs[lockKey].length}`, LOG_LEVEL.VERBOSE);
             let nextProc = null;
             nextProc = pendingProcs[lockKey].shift();
+            notifyLock();
             if (nextProc) {
                 // left some
                 nextProc()
@@ -145,6 +168,7 @@ export function runWithLock<T>(key: unknown, ignoreWhenRunning: boolean, proc: (
                     .finally(() => {
                         if (pendingProcs && lockKey in pendingProcs && pendingProcs[lockKey].length == 0) {
                             delete pendingProcs[lockKey];
+                            notifyLock();
                         }
                         queueMicrotask(() => {
                             handleNextProcs();
@@ -153,6 +177,7 @@ export function runWithLock<T>(key: unknown, ignoreWhenRunning: boolean, proc: (
             } else {
                 if (pendingProcs && lockKey in pendingProcs && pendingProcs[lockKey].length == 0) {
                     delete pendingProcs[lockKey];
+                    notifyLock();
                 }
             }
         }
@@ -189,10 +214,12 @@ export function runWithLock<T>(key: unknown, ignoreWhenRunning: boolean, proc: (
             });
 
         pendingProcs[lockKey].push(subproc);
+        notifyLock();
         // Logger(`Lock:${lockKey}:queud:left${pendingProcs[lockKey].length}`, LOG_LEVEL.VERBOSE);
         return responder;
     } else {
         runningProcs.push(lockKey);
+        notifyLock();
         // Logger(`Lock:${lockKey}:aqquired`, LOG_LEVEL.VERBOSE);
         return new Promise((res, rej) => {
             proc()

src/utils_couchdb.ts

@@ -8,11 +8,54 @@ export const isValidRemoteCouchDBURI = (uri: string): boolean => {
     if (uri.startsWith("http://")) return true;
     return false;
 };
+let last_post_successed = false;
+export const getLastPostFailedBySize = () => {
+    return !last_post_successed;
+};
 export const connectRemoteCouchDB = async (uri: string, auth: { username: string; password: string }): Promise<string | { db: PouchDB.Database<EntryDoc>; info: PouchDB.Core.DatabaseInfo }> => {
     if (!isValidRemoteCouchDBURI(uri)) return "Remote URI is not valid";
-    const db: PouchDB.Database<EntryDoc> = new PouchDB<EntryDoc>(uri, {
+    const conf: PouchDB.HttpAdapter.HttpAdapterConfiguration = {
+        adapter: "http",
         auth,
-    });
+        fetch: async function (url: string | Request, opts: RequestInit) {
+            let size_ok = true;
+            let size = "";
+            const localURL = url.toString().substring(uri.length);
+            const method = opts.method ?? "GET";
+            if (opts.body) {
+                const opts_length = opts.body.toString().length;
+                if (opts_length > 1024 * 1024 * 10) {
+                    // over 10MB
+                    size_ok = false;
+                    if (uri.contains(".cloudantnosqldb.")) {
+                        last_post_successed = false;
+                        Logger("This request should fail on IBM Cloudant.", LOG_LEVEL.VERBOSE);
+                        throw new Error("This request should fail on IBM Cloudant.");
+                    }
+                }
+                size = ` (${opts_length})`;
+            }
+            try {
+                const responce: Response = await fetch(url, opts);
+                if (method == "POST" || method == "PUT") {
+                    last_post_successed = responce.ok;
+                } else {
+                    last_post_successed = true;
+                }
+                Logger(`HTTP:${method}${size} to:${localURL} -> ${responce.status}`, LOG_LEVEL.VERBOSE);
+                return responce;
+            } catch (ex) {
+                Logger(`HTTP:${method}${size} to:${localURL} -> failed`, LOG_LEVEL.VERBOSE);
+                if (!size_ok && (method == "POST" || method == "PUT")) {
+                    last_post_successed = false;
+                }
+                Logger(ex);
+                throw ex;
+            }
+            // return await fetch(url, opts);
+        },
+    };
+    const db: PouchDB.Database<EntryDoc> = new PouchDB<EntryDoc>(uri, conf);
     try {
         const info = await db.info();
         return { db: db, info: info };