From 59374a6024184b47f32d8c3912715bcd25c82b45 Mon Sep 17 00:00:00 2001 From: Will Pike <6687499+pike00@users.noreply.github.com> Date: Thu, 30 Jul 2026 22:57:46 -0500 Subject: [PATCH] Implement inline deletion reconciliation Adds waitOfflineScanComplete to PeerStorage using Promise.withResolvers to wait for Deno.watchFs or chokidar offline scans to finish. Adds reconcileDiskDeletions to Hub which waits for storage peers to scan, enumerates docs in CouchDB peers, and deletes any docs from CouchDB that are missing on the corresponding local disk. Adds test for the reconciliation behavior in Patch4.test.ts. --- Hub.ts | 184 ++++++++----- PeerStorage.ts | 628 +++++++++++++++++++++++-------------------- tests/Patch4.test.ts | 40 +++ 3 files changed, 492 insertions(+), 360 deletions(-) create mode 100644 tests/Patch4.test.ts diff --git a/Hub.ts b/Hub.ts index 645f8d5..3779c23 100644 --- a/Hub.ts +++ b/Hub.ts @@ -3,76 +3,132 @@ import { Peer, PeerHealth } from "./Peer.ts"; import { PeerStorage } from "./PeerStorage.ts"; import { PeerCouchDB } from "./PeerCouchDB.ts"; - export class Hub { - conf: Config; - peers = [] as Peer[]; - constructor(conf: Config) { - this.conf = conf; + conf: Config; + peers = [] as Peer[]; + constructor(conf: Config) { + this.conf = conf; + } + // Aggregate peer health for the heartbeat. `ok` = every peer syncing (also + // false if no peers were constructed). `restartWorthy` = any peer judges itself + // restart-worthy (was healthy, now persistently failing while its backend is + // up) — see Peer.probeHealth. + async healthProbe(): Promise< + { ok: boolean; restartWorthy: boolean; peers: PeerHealth[] } + > { + const peers = await Promise.all(this.peers.map((p) => p.probeHealth())); + const ok = peers.length > 0 && peers.every((p) => p.ok); + const restartWorthy = peers.some((p) => p.restartWorthy); + return { ok, restartWorthy, peers }; + } + async reconcileDiskDeletions() { + const storagePeers = this.peers.filter((p) => + p.config.type === "storage" + ) as PeerStorage[]; + await Promise.all(storagePeers.map((p) => p.waitOfflineScanComplete())); + + for (const cp of this.peers) { + if (cp.config.type !== "couchdb") continue; + const matchingStoragePeers = storagePeers.filter((p) => + (p.config.group ?? "") === (cp.config.group ?? "") + ); + if (matchingStoragePeers.length === 0) continue; + + const peerCouchDb = cp as PeerCouchDB; + for await ( + const doc of peerCouchDb.man.enumerateAllNormalDocs({ metaOnly: true }) + ) { + if (doc.deleted || (doc as any)._deleted) continue; + + let path = (doc as any).path.substring( + peerCouchDb.toLocalPath("").length, + ); + if (path.startsWith("/")) path = path.substring(1); + if (path.startsWith("i:")) path = path.substring(2); + + const globalPath = peerCouchDb.toGlobalPath(path); + let existsOnDisk = false; + for (const sp of matchingStoragePeers) { + try { + if (await sp.get(globalPath) !== false) { + existsOnDisk = true; + break; + } + } catch (e) { + // Fallback for not found + } + } + + if (!existsOnDisk) { + await peerCouchDb.delete(globalPath); + } + } } - // Aggregate peer health for the heartbeat. `ok` = every peer syncing (also - // false if no peers were constructed). `restartWorthy` = any peer judges itself - // restart-worthy (was healthy, now persistently failing while its backend is - // up) — see Peer.probeHealth. - async healthProbe(): Promise<{ ok: boolean; restartWorthy: boolean; peers: PeerHealth[] }> { - const peers = await Promise.all(this.peers.map((p) => p.probeHealth())); - const ok = peers.length > 0 && peers.every((p) => p.ok); - const restartWorthy = peers.some((p) => p.restartWorthy); - return { ok, restartWorthy, peers }; + } + + start() { + for (const p of this.peers) { + p.stop(); + } + this.peers = []; + for (const peer of this.conf.peers) { + if (peer.type == "couchdb") { + const p = new PeerCouchDB(peer, this.dispatch.bind(this)); + this.peers.push(p); + } else if (peer.type == "storage") { + const p = new PeerStorage(peer, this.dispatch.bind(this)); + this.peers.push(p); + } else { + throw new Error( + `Unexpected Peer type: ${(peer as any)?.name} - ${ + (peer as any)?.type + }`, + ); + } } - start() { - for (const p of this.peers) { - p.stop(); + // Initialize couchdb peers FIRST and await them, then start storage peers. + // Otherwise a storage peer's offline scan can push to a couchdb peer before its + // DB managers are initialized (initializeDatabase), causing + // "Cannot read properties of undefined (reading 'getDBEntryMeta')". + (async () => { + for (const p of this.peers) { + if (p.config.type === "couchdb") { + await p.start().catch((e) => { + console.error(`[Hub] peer "${p.config.name}" start() failed:`, e); + }); } - this.peers = []; - for (const peer of this.conf.peers) { - if (peer.type == "couchdb") { - const p = new PeerCouchDB(peer, this.dispatch.bind(this)); - this.peers.push(p); - } else if (peer.type == "storage") { - const p = new PeerStorage(peer, this.dispatch.bind(this)); - this.peers.push(p); - } else { - throw new Error(`Unexpected Peer type: ${(peer as any)?.name} - ${(peer as any)?.type}`); - } + } + for (const p of this.peers) { + if (p.config.type !== "couchdb") { + p.start().catch((e) => { + console.error(`[Hub] peer "${p.config.name}" start() failed:`, e); + }); } - // Initialize couchdb peers FIRST and await them, then start storage peers. - // Otherwise a storage peer's offline scan can push to a couchdb peer before its - // DB managers are initialized (initializeDatabase), causing - // "Cannot read properties of undefined (reading 'getDBEntryMeta')". - (async () => { - for (const p of this.peers) { - if (p.config.type === "couchdb") { - await p.start().catch((e) => { - console.error(`[Hub] peer "${p.config.name}" start() failed:`, e); - }); - } - } - for (const p of this.peers) { - if (p.config.type !== "couchdb") { - p.start().catch((e) => { - console.error(`[Hub] peer "${p.config.name}" start() failed:`, e); - }); - } - } - })(); - } + } + this.reconcileDiskDeletions().catch((e) => + console.error("[Hub] reconcileDiskDeletions failed:", e) + ); + })(); + } - async dispatch(source: Peer, path: string, data: FileData | false) { - for (const peer of this.peers) { - if (peer !== source && (source.config.group ?? "") === (peer.config.group ?? "")) { - let ret = false; - if (data === false) { - ret = await peer.delete(path); - } else { - ret = await peer.put(path, data); - } - if (ret) { - // Logger(` ${data === false ? "-x->" : "--->"} ${peer.config.name} ${path} `) - } else { - // Logger(` ${peer.config.name} ignored ${path} `) - } - } + async dispatch(source: Peer, path: string, data: FileData | false) { + for (const peer of this.peers) { + if ( + peer !== source && + (source.config.group ?? "") === (peer.config.group ?? "") + ) { + let ret = false; + if (data === false) { + ret = await peer.delete(path); + } else { + ret = await peer.put(path, data); + } + if (ret) { + // Logger(` ${data === false ? "-x->" : "--->"} ${peer.config.name} ${path} `) + } else { + // Logger(` ${peer.config.name} ignored ${path} `) } + } } + } } diff --git a/PeerStorage.ts b/PeerStorage.ts index 9dbd088..3cc0d27 100644 --- a/PeerStorage.ts +++ b/PeerStorage.ts @@ -1,338 +1,374 @@ -import { LOG_LEVEL_INFO, LOG_LEVEL_NOTICE, LOG_LEVEL_VERBOSE } from "./lib/src/common/types.ts"; -import { PeerStorageConf, FileData } from "./types.ts"; +import { + LOG_LEVEL_INFO, + LOG_LEVEL_NOTICE, + LOG_LEVEL_VERBOSE, +} from "./lib/src/common/types.ts"; +import { FileData, PeerStorageConf } from "./types.ts"; import { Logger } from "./lib/src/common/logger.ts"; import { delay, getDocData } from "./lib/src/common/utils.ts"; import { isPlainText } from "./lib/src/string_and_binary/path.ts"; -import { parse, format, relative, dirname, resolve } from "@std/path"; -import { format as posixFormat, parse as posixParse } from "@std/path/posix" +import { dirname, format, parse, relative, resolve } from "@std/path"; +import { format as posixFormat, parse as posixParse } from "@std/path/posix"; import { scheduleOnceIfDuplicated } from "octagonal-wheels/concurrency/lock"; import { DispatchFun, Peer, PeerHealth } from "./Peer.ts"; import chokidar from "chokidar"; -import { walk } from 'fs/walk'; +import { walk } from "fs/walk"; import { scheduleTask } from "octagonal-wheels/concurrency/task"; export class PeerStorage extends Peer { - declare config: PeerStorageConf; + declare config: PeerStorageConf; + private _offlineScanFinished = Promise.withResolvers(); + constructor(conf: PeerStorageConf, dispatcher: DispatchFun) { + super(conf, dispatcher); + } - constructor(conf: PeerStorageConf, dispatcher: DispatchFun) { - super(conf, dispatcher); + async delete(pathSrc: string): Promise { + const lp = this.toLocalPath(pathSrc); + const path = this.toStoragePath(lp); + if (await this.isRepeating(lp, false)) { + return false; } - - async delete(pathSrc: string): Promise { - const lp = this.toLocalPath(pathSrc); - const path = this.toStoragePath(lp); - if (await this.isRepeating(lp, false)) { - return false; - } - try { - await Deno.remove(path); - this.receiveLog(` ${path} deleted`); - } catch (ex) { - this.receiveLog(` ${path} delete failed`, LOG_LEVEL_NOTICE); - Logger(ex, LOG_LEVEL_VERBOSE); - return false; - } - this.runScript(path, true); - return true; + try { + await Deno.remove(path); + this.receiveLog(` ${path} deleted`); + } catch (ex) { + this.receiveLog(` ${path} delete failed`, LOG_LEVEL_NOTICE); + Logger(ex, LOG_LEVEL_VERBOSE); + return false; } - async put(pathSrc: string, data: FileData): Promise { - const lp = this.toLocalPath(pathSrc); - const path = this.toStoragePath(lp); - if (await this.isRepeating(lp, data)) { - this.receiveLog(`${lp} save repeating`); - return false; - } - try { - const dirName = dirname(path); - try { - await Deno.mkdir(dirName, { recursive: true }); - } catch (ex) { - // While recursive is true, mkdir will not raise the `AlreadyExist`. - console.log(ex); - } - const fp = await Deno.open(path, { read: true, write: true, create: true }); - if (data.data instanceof Uint8Array) { - const writtensize = await fp.write(data.data); - await fp.truncate(writtensize); - } else { - const writtensize = await fp.write(new TextEncoder().encode(getDocData(data.data))); - await fp.truncate(writtensize); - } - await fp.utime(new Date(data.mtime), new Date(data.mtime)); - fp.close(); - this.receiveLog(`${lp} saved`); - await this.writeFileStat(pathSrc); - this.runScript(path, false); - return true; - } catch (ex) { - Logger(ex, LOG_LEVEL_INFO); - this.receiveLog(`${lp} save failed`); - return false; - } + this.runScript(path, true); + return true; + } + async put(pathSrc: string, data: FileData): Promise { + const lp = this.toLocalPath(pathSrc); + const path = this.toStoragePath(lp); + if (await this.isRepeating(lp, data)) { + this.receiveLog(`${lp} save repeating`); + return false; } + try { + const dirName = dirname(path); + try { + await Deno.mkdir(dirName, { recursive: true }); + } catch (ex) { + // While recursive is true, mkdir will not raise the `AlreadyExist`. + console.log(ex); + } + const fp = await Deno.open(path, { + read: true, + write: true, + create: true, + }); + if (data.data instanceof Uint8Array) { + const writtensize = await fp.write(data.data); + await fp.truncate(writtensize); + } else { + const writtensize = await fp.write( + new TextEncoder().encode(getDocData(data.data)), + ); + await fp.truncate(writtensize); + } + await fp.utime(new Date(data.mtime), new Date(data.mtime)); + fp.close(); + this.receiveLog(`${lp} saved`); + await this.writeFileStat(pathSrc); + this.runScript(path, false); + return true; + } catch (ex) { + Logger(ex, LOG_LEVEL_INFO); + this.receiveLog(`${lp} save failed`); + return false; + } + } - async runScript(filename: string, isDeleted: boolean): Promise { - if (!this.config.processor) return false; - if (!this.config.processor.cmd) return false; - - // const result = []; - try { - // const startDate = new Date(); - const cmd = this.config.processor.cmd; - const mode = isDeleted ? "deleted" : "modified"; - const args = this.config.processor.args.map(e => { - if (e == "$filename") return filename; - if (e == "$mode") return mode; - return e - }); - // const dateStr = startDate.toLocaleString(); - const scriptLineMessage = `Script: called ${cmd} with args ${JSON.stringify(args)}`; - this.normalLog(`Processor : ${scriptLineMessage}`) - const command = new Deno.Command( - cmd, { - args: args, - cwd: ".", - env: { - filename: filename, - mode: mode - } - }); - // const start = performance.now(); - const { code, stdout, stderr } = await command.output(); - // const end = performance.now(); - const stdoutText = new TextDecoder().decode(stdout); - const stderrText = new TextDecoder().decode(stderr); - // result.push(`# Processor called: ${dateStr}\n`); - // result.push(`command: \`${scriptLineMessage}\``); - if (code === 0) { - this.normalLog("Processor called: Performed successfully.") - // result.push("Processor called: Performed successfully.") - this.normalLog(stdoutText); - } else { - this.normalLog("Processor called: Performed but with some errors.") - // result.push("Processor called: Performed but with some errors.") - this.normalLog(stderrText, LOG_LEVEL_NOTICE); - } - // result.push(`\n- Spent ${Math.ceil(end - start) / 1000} ms`); - // result.push("## --STDOUT--\n") - // result.push("```\n" + stdoutText + "\n```"); - // result.push("## --STDERR--n") - // result.push("```\n" + stderrText + "\n```"); - // const strResult = result.join("\n"); - return true; - } catch (ex) { - this.normalLog("Processor: Error on processing");; - // this.normalLog(ex); - this.normalLog(JSON.stringify(ex, null, 2)); - return false; - } + async runScript(filename: string, isDeleted: boolean): Promise { + if (!this.config.processor) return false; + if (!this.config.processor.cmd) return false; + // const result = []; + try { + // const startDate = new Date(); + const cmd = this.config.processor.cmd; + const mode = isDeleted ? "deleted" : "modified"; + const args = this.config.processor.args.map((e) => { + if (e == "$filename") return filename; + if (e == "$mode") return mode; + return e; + }); + // const dateStr = startDate.toLocaleString(); + const scriptLineMessage = `Script: called ${cmd} with args ${ + JSON.stringify(args) + }`; + this.normalLog(`Processor : ${scriptLineMessage}`); + const command = new Deno.Command( + cmd, + { + args: args, + cwd: ".", + env: { + filename: filename, + mode: mode, + }, + }, + ); + // const start = performance.now(); + const { code, stdout, stderr } = await command.output(); + // const end = performance.now(); + const stdoutText = new TextDecoder().decode(stdout); + const stderrText = new TextDecoder().decode(stderr); + // result.push(`# Processor called: ${dateStr}\n`); + // result.push(`command: \`${scriptLineMessage}\``); + if (code === 0) { + this.normalLog("Processor called: Performed successfully."); + // result.push("Processor called: Performed successfully.") + this.normalLog(stdoutText); + } else { + this.normalLog("Processor called: Performed but with some errors."); + // result.push("Processor called: Performed but with some errors.") + this.normalLog(stderrText, LOG_LEVEL_NOTICE); + } + // result.push(`\n- Spent ${Math.ceil(end - start) / 1000} ms`); + // result.push("## --STDOUT--\n") + // result.push("```\n" + stdoutText + "\n```"); + // result.push("## --STDERR--n") + // result.push("```\n" + stderrText + "\n```"); + // const strResult = result.join("\n"); + return true; + } catch (ex) { + this.normalLog("Processor: Error on processing"); + // this.normalLog(ex); + this.normalLog(JSON.stringify(ex, null, 2)); + return false; } + } - async get(pathSrc: string): Promise { - const lp = this.toLocalPath(pathSrc); - const path = this.toStoragePath(lp); - const stat = await Deno.stat(path); - if (!stat.isFile) { - return false; - } - const ret: FileData = { - ctime: stat.mtime?.getTime() ?? 0, - mtime: stat.mtime?.getTime() ?? 0, - size: stat.size, - data: [], - }; - if (isPlainText(path)) { - ret.data = [await Deno.readTextFile(path)]; - } else { - ret.data = await Deno.readFile(path); - } - return ret; + async get(pathSrc: string): Promise { + const lp = this.toLocalPath(pathSrc); + const path = this.toStoragePath(lp); + const stat = await Deno.stat(path); + if (!stat.isFile) { + return false; } - watcher?: chokidar.FSWatcher; + const ret: FileData = { + ctime: stat.mtime?.getTime() ?? 0, + mtime: stat.mtime?.getTime() ?? 0, + size: stat.size, + data: [], + }; + if (isPlainText(path)) { + ret.data = [await Deno.readTextFile(path)]; + } else { + ret.data = await Deno.readFile(path); + } + return ret; + } - async dispatch(pathSrc: string) { - const lP = this.toStoragePath(this.toLocalPath(".")); - const path = this.toPosixPath(relative(lP, pathSrc)); + waitOfflineScanComplete(): Promise { + return this._offlineScanFinished.promise; + } - const data = await this.get(path); + watcher?: chokidar.FSWatcher; - if (data === false) return; + async dispatch(pathSrc: string) { + const lP = this.toStoragePath(this.toLocalPath(".")); + const path = this.toPosixPath(relative(lP, pathSrc)); - scheduleOnceIfDuplicated(pathSrc, async () => { - // console.log(data); - await this.writeFileStat(path); - await delay(250); - if (!await this.isRepeating(path, data)) { - this.sendLog(`${path} change detected`); - await this.dispatchToHub(this, this.toGlobalPath(path), data); - } - // else { - // this.sendLog(`${path} change repeating detected`); - // } - }); - } - async dispatchDeleted(pathSrc: string) { - const lP = this.toStoragePath(this.toLocalPath(".")); - const path = this.toPosixPath(relative(lP, pathSrc)); - await scheduleOnceIfDuplicated(pathSrc, async () => { - await delay(250); - if (!await this.isRepeating(path, false)) { - this.sendLog(`${path} delete detected`); - await this.dispatchToHub(this, this.toGlobalPath(path), false); - } - }); + const data = await this.get(path); - } + if (data === false) return; - toPosixPath(path: string) { - const ret = posixFormat(parse(path)); - // this.debugLog(`**TOPOSIX ${path} -> ${ret}`) - return ret; - } - toStoragePath(path: string) { - const ret = resolve(format(posixParse(path))); - // this.debugLog(`**TOSTORAGE ${path} -> ${ret}`) - return ret; - } + scheduleOnceIfDuplicated(pathSrc, async () => { + // console.log(data); + await this.writeFileStat(path); + await delay(250); + if (!await this.isRepeating(path, data)) { + this.sendLog(`${path} change detected`); + await this.dispatchToHub(this, this.toGlobalPath(path), data); + } + // else { + // this.sendLog(`${path} change repeating detected`); + // } + }); + } + async dispatchDeleted(pathSrc: string) { + const lP = this.toStoragePath(this.toLocalPath(".")); + const path = this.toPosixPath(relative(lP, pathSrc)); + await scheduleOnceIfDuplicated(pathSrc, async () => { + await delay(250); + if (!await this.isRepeating(path, false)) { + this.sendLog(`${path} delete detected`); + await this.dispatchToHub(this, this.toGlobalPath(path), false); + } + }); + } - async writeFileStat(pathSrc: string, statSrc?: Deno.FileInfo) { - const lp = this.toLocalPath(pathSrc); - const key = `file-stat-${lp}`; - const path = this.toStoragePath(lp); - const stat = statSrc ?? await Deno.stat(path); - if (!stat.isFile) { - return false; - } - const fileStat = `${stat.mtime?.getTime() ?? 0}-${stat.size}`; - this.setSetting(key, fileStat); + toPosixPath(path: string) { + const ret = posixFormat(parse(path)); + // this.debugLog(`**TOPOSIX ${path} -> ${ret}`) + return ret; + } + toStoragePath(path: string) { + const ret = resolve(format(posixParse(path))); + // this.debugLog(`**TOSTORAGE ${path} -> ${ret}`) + return ret; + } + + async writeFileStat(pathSrc: string, statSrc?: Deno.FileInfo) { + const lp = this.toLocalPath(pathSrc); + const key = `file-stat-${lp}`; + const path = this.toStoragePath(lp); + const stat = statSrc ?? await Deno.stat(path); + if (!stat.isFile) { + return false; } + const fileStat = `${stat.mtime?.getTime() ?? 0}-${stat.size}`; + this.setSetting(key, fileStat); + } - async isChanged(pathSrc: string) { - const lp = this.toLocalPath(pathSrc); - const key = `file-stat-${lp}`; - const last = this.getSetting(key); - // console.log(`R:${key}`); - // console.log(`RV:${last}`); + async isChanged(pathSrc: string) { + const lp = this.toLocalPath(pathSrc); + const key = `file-stat-${lp}`; + const last = this.getSetting(key); + // console.log(`R:${key}`); + // console.log(`RV:${last}`); - const path = this.toStoragePath(lp); - const stat = await Deno.stat(path); - if (!stat.isFile) { - return false; - } - if (!last) return true; - const fileStat = `${stat.mtime?.getTime() ?? 0}-${stat.size}`; - // console.log(`RVX:${fileStat}`); - if (last !== fileStat) return true; - return false; + const path = this.toStoragePath(lp); + const stat = await Deno.stat(path); + if (!stat.isFile) { + return false; } - watcherDeno?: Deno.FsWatcher; + if (!last) return true; + const fileStat = `${stat.mtime?.getTime() ?? 0}-${stat.size}`; + // console.log(`RVX:${fileStat}`); + if (last !== fileStat) return true; + return false; + } + watcherDeno?: Deno.FsWatcher; - processFile(event: Deno.FsEvent) { - for (const path of event.paths) { - const key = `${event.kind}-${path}`; - // const key = path; - scheduleTask(key, 100, async () => { - const existence = await Deno.stat(path).catch(() => null); - if (existence) { - if (existence.isFile) { - await this.dispatch(path); - } - } else { - await this.dispatchDeleted(path); - } - }); + processFile(event: Deno.FsEvent) { + for (const path of event.paths) { + const key = `${event.kind}-${path}`; + // const key = path; + scheduleTask(key, 100, async () => { + const existence = await Deno.stat(path).catch(() => null); + if (existence) { + if (existence.isFile) { + await this.dispatch(path); + } + } else { + await this.dispatchDeleted(path); } + }); } + } - - - async startDenoFsWatch(): Promise { - if (this.watcherDeno) { - this.watcherDeno.close(); - this.watcherDeno = undefined; - } - const lP = this.toStoragePath(this.toLocalPath(".")); - this.normalLog(`Scan offline changes: ${this.config.scanOfflineChanges ? "Enabled, now starting..." : "Disabled"}`); - if (this.config.scanOfflineChanges) { - for await (const entry of walk(lP)) { - if (entry.isFile) { - const ePath = this.toPosixPath(relative(this.toLocalPath("."), entry.path)); - if (await this.isChanged(ePath)) { - this.debugLog(`Offline changes detected: ${ePath}`); - await this.dispatch(entry.path); - } - } - } - } - this.watcherDeno = Deno.watchFs(lP, - { - recursive: true, - }); - - for await (const event of this.watcherDeno) { - this.processFile(event); - } - + async startDenoFsWatch(): Promise { + if (this.watcherDeno) { + this.watcherDeno.close(); + this.watcherDeno = undefined; } - async start() { - // For addressing Deno's and chokidar's compatibility issues (especially on Windows), we use Deno's fs watcher as the primary watcher. - if (!this.config.useChokidar) { - await this.startDenoFsWatch(); - return; - } - - if (this.watcher) { - this.watcher.close(); - this.watcher = undefined; + const lP = this.toStoragePath(this.toLocalPath(".")); + this.normalLog( + `Scan offline changes: ${ + this.config.scanOfflineChanges ? "Enabled, now starting..." : "Disabled" + }`, + ); + if (this.config.scanOfflineChanges) { + for await (const entry of walk(lP)) { + if (entry.isFile) { + const ePath = this.toPosixPath( + relative(this.toLocalPath("."), entry.path), + ); + if (await this.isChanged(ePath)) { + this.debugLog(`Offline changes detected: ${ePath}`); + await this.dispatch(entry.path); + } } - const lP = this.toStoragePath(this.toLocalPath(".")); - this.normalLog(`Scan offline changes: ${this.config.scanOfflineChanges ? "Enabled, now starting..." : "Disabled"}`); - this.watcher = chokidar.watch(lP, - { - ignoreInitial: !this.config.scanOfflineChanges, - awaitWriteFinish: { - stabilityThreshold: 500, - }, - }); + } + this._offlineScanFinished.resolve(); + } else { + this._offlineScanFinished.resolve(); + } + this.watcherDeno = Deno.watchFs(lP, { + recursive: true, + }); - this.watcher.on("change", async (path) => { - const ePath = this.toPosixPath(relative(this.toLocalPath("."), path)); - if (!await this.isChanged(ePath)) { - // this.debugLog(`Not changed: ${ePath}`); - } else { - this.debugLog(`Changes detected: ${ePath}`); - await this.dispatch(path); - } - }) - this.watcher.on("add", async (path) => { - const ePath = this.toPosixPath(relative(this.toLocalPath("."), path)); - if (!await this.isChanged(ePath)) { - // this.debugLog(`Not changed: ${ePath}`); - } else { - this.debugLog(`New detected: ${ePath}`); - await this.dispatch(path); - } - }) - this.watcher.on("unlink", async (path) => { - const ePath = this.toPosixPath(relative(this.toLocalPath("."), path)); - this.debugLog(`Unlink detected: ${ePath}`); - await this.dispatchDeleted(path) - }) + for await (const event of this.watcherDeno) { + this.processFile(event); } - async stop() { - this.watcher?.close(); - this.watcherDeno?.close(); - this.watcherDeno = undefined; - return await Promise.resolve(); + } + async start() { + // For addressing Deno's and chokidar's compatibility issues (especially on Windows), we use Deno's fs watcher as the primary watcher. + if (!this.config.useChokidar) { + await this.startDenoFsWatch(); + return; } - override health(): PeerHealth { - const ok = !!(this.watcherDeno || this.watcher); - // No remote backend (backendUp always true). A storage peer still doing its - // initial offline scan is "starting", not yet healthy, so the base restart - // logic won't flag it — only a watcher that dies after being healthy counts. - return { name: this.config.name, type: "storage", ok, detail: ok ? "watching" : "starting", backendUp: true, restartWorthy: false }; + + if (this.watcher) { + this.watcher.close(); + this.watcher = undefined; } + const lP = this.toStoragePath(this.toLocalPath(".")); + this.normalLog( + `Scan offline changes: ${ + this.config.scanOfflineChanges ? "Enabled, now starting..." : "Disabled" + }`, + ); + this.watcher = chokidar.watch(lP, { + ignoreInitial: !this.config.scanOfflineChanges, + awaitWriteFinish: { + stabilityThreshold: 500, + }, + }); + + this.watcher.on("ready", () => { + this._offlineScanFinished.resolve(); + }); + + this.watcher.on("change", async (path) => { + const ePath = this.toPosixPath(relative(this.toLocalPath("."), path)); + if (!await this.isChanged(ePath)) { + // this.debugLog(`Not changed: ${ePath}`); + } else { + this.debugLog(`Changes detected: ${ePath}`); + await this.dispatch(path); + } + }); + this.watcher.on("add", async (path) => { + const ePath = this.toPosixPath(relative(this.toLocalPath("."), path)); + if (!await this.isChanged(ePath)) { + // this.debugLog(`Not changed: ${ePath}`); + } else { + this.debugLog(`New detected: ${ePath}`); + await this.dispatch(path); + } + }); + this.watcher.on("unlink", async (path) => { + const ePath = this.toPosixPath(relative(this.toLocalPath("."), path)); + this.debugLog(`Unlink detected: ${ePath}`); + await this.dispatchDeleted(path); + }); + } + async stop() { + this.watcher?.close(); + this.watcherDeno?.close(); + this.watcherDeno = undefined; + return await Promise.resolve(); + } + override health(): PeerHealth { + const ok = !!(this.watcherDeno || this.watcher); + // No remote backend (backendUp always true). A storage peer still doing its + // initial offline scan is "starting", not yet healthy, so the base restart + // logic won't flag it — only a watcher that dies after being healthy counts. + return { + name: this.config.name, + type: "storage", + ok, + detail: ok ? "watching" : "starting", + backendUp: true, + restartWorthy: false, + }; + } } diff --git a/tests/Patch4.test.ts b/tests/Patch4.test.ts new file mode 100644 index 0000000..d341b59 --- /dev/null +++ b/tests/Patch4.test.ts @@ -0,0 +1,40 @@ +import { assertEquals } from "@std/assert"; +import { Hub } from "../Hub.ts"; +import { PeerStorage } from "../PeerStorage.ts"; +import { PeerCouchDB } from "../PeerCouchDB.ts"; + +Deno.test("reconcileDiskDeletions deletes missing CouchDB doc", async () => { + const hub = new Hub({ + peers: [ + { type: "couchdb", name: "test-db", group: "test-group", url: "", database: "", baseDir: "" }, + { type: "storage", name: "test-storage", group: "test-group", baseDir: "" } + ] + }); + + const cp = new PeerCouchDB(hub.conf.peers[0] as any, async () => {}); + const sp = new PeerStorage(hub.conf.peers[1] as any, async () => {}); + + let deleteCalled = false; + cp.delete = async (path) => { + if (path === "test-file.md") deleteCalled = true; + return true; + }; + + // Mock man enumerateAllNormalDocs + cp.man = { + enumerateAllNormalDocs: async function*(opt: any) { + yield { path: "test-file.md", deleted: false }; + } + } as any; + + sp.get = async () => false; + + // Resolve offline scan automatically for the test + sp.waitOfflineScanComplete = async () => Promise.resolve(); + + hub.peers = [cp, sp]; + + await hub.reconcileDiskDeletions(); + + assertEquals(deleteCalled, true); +});