fix(watchlist): stale-read clobber, concurrency guard, leaked timeout, quiet catch-up notices
- SyncDeps.listNotes now returns paths only; new readNote(path) reads content immediately before parse/diff/write per note, closing the window where an edit made between the initial scan and a later write got silently overwritten. resolveMissingIds updated to match. - WatchlistController gains a syncing flag shared by syncNow and resolveMissingIds so overlapping invocations short-circuit instead of racing on the same vault notes. - main.ts registers cleanup for the 30s startup catch-up setTimeout so it doesn't fire after unload. - syncNow takes a quiet flag (used by the scheduled catch-up path) so zero-change scheduled syncs don't spam a Notice; manual commands are unaffected.
This commit is contained in:
parent
b77a189848
commit
a4ba1b8e42
5 changed files with 208 additions and 72 deletions
|
|
@ -65,7 +65,8 @@ export default class MediaDbPlugin extends Plugin {
|
|||
|
||||
this.watchlist = new WatchlistController(this);
|
||||
// catch-up shortly after startup (let vault index settle), then hourly due-check
|
||||
window.setTimeout(() => void this.watchlist.maybeCatchUp(), 30_000);
|
||||
const catchUpTimeout = window.setTimeout(() => void this.watchlist.maybeCatchUp(), 30_000);
|
||||
this.register(() => window.clearTimeout(catchUpTimeout));
|
||||
this.registerInterval(window.setInterval(() => void this.watchlist.maybeCatchUp(), 3600_000));
|
||||
}
|
||||
|
||||
|
|
@ -92,7 +93,7 @@ export default class MediaDbPlugin extends Plugin {
|
|||
|
||||
private registerRibbonAndFileMenu(): void {
|
||||
const ribbonIconEl = this.addRibbonIcon('database', 'Add new Media DB entry', () => this.entryHelper.createEntryWithAdvancedSearchModal());
|
||||
ribbonIconEl.addClass('obsidian-media-db-plugin-ribbon-class');
|
||||
ribbonIconEl.addClass('media-db-sync-ribbon-class');
|
||||
|
||||
this.registerEvent(
|
||||
this.app.workspace.on('file-menu', (menu, file) => {
|
||||
|
|
|
|||
|
|
@ -3,7 +3,8 @@ import { buildRecord } from 'packages/obsidian/src/watchlist/build';
|
|||
import { renderNote } from 'packages/obsidian/src/watchlist/render';
|
||||
|
||||
export interface SyncDeps {
|
||||
listNotes(): Promise<{ path: string; content: string }[]>;
|
||||
listNotes(): Promise<{ path: string }[]>;
|
||||
readNote(path: string): Promise<string>;
|
||||
writeNote(path: string, content: string): Promise<void>;
|
||||
fetchDetail(tmdbId: string, isMovie: boolean): Promise<any>;
|
||||
sleep(ms: number): Promise<void>;
|
||||
|
|
@ -64,17 +65,19 @@ export async function syncFolder(deps: SyncDeps, opts: SyncOptions = {}): Promis
|
|||
const notes = await deps.listNotes();
|
||||
for (const note of notes) {
|
||||
report.scanned++;
|
||||
const { frontmatter, body } = parseNote(note.content);
|
||||
const ref = noteTmdbRef(frontmatter);
|
||||
if (!ref) {
|
||||
report.skippedNoId++;
|
||||
continue;
|
||||
}
|
||||
if (!opts.full && !isActive(frontmatter)) {
|
||||
report.skippedStatic++;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
// Read immediately before parse/diff so mid-sync edits aren't clobbered by a stale snapshot.
|
||||
const content = await deps.readNote(note.path);
|
||||
const { frontmatter, body } = parseNote(content);
|
||||
const ref = noteTmdbRef(frontmatter);
|
||||
if (!ref) {
|
||||
report.skippedNoId++;
|
||||
continue;
|
||||
}
|
||||
if (!opts.full && !isActive(frontmatter)) {
|
||||
report.skippedStatic++;
|
||||
continue;
|
||||
}
|
||||
const detail: any = await withRateLimitRetry(() => deps.fetchDetail(ref.tmdbId, ref.isMovie), deps.sleep);
|
||||
const record = buildRecord(detail, ref.isMovie, frontmatter);
|
||||
const rendered = renderNote(record, extractMyNotes(body));
|
||||
|
|
@ -82,7 +85,7 @@ export async function syncFolder(deps: SyncDeps, opts: SyncOptions = {}): Promis
|
|||
if (strip(frontmatter['watch_status']) === 'Watched' && record.watchStatus === 'Unwatched') {
|
||||
report.flipped.push(note.path);
|
||||
}
|
||||
if (rendered !== note.content) {
|
||||
if (rendered !== content) {
|
||||
report.written++;
|
||||
if (!opts.dryRun) await deps.writeNote(note.path, rendered);
|
||||
deps.log(`${opts.dryRun ? '[dry] ' : ''}updated ${note.path}`);
|
||||
|
|
|
|||
|
|
@ -25,7 +25,13 @@ function patchFrontmatter(content: string, tmdbId: string, mediaType: string): s
|
|||
return `---\ntype: watchlist_item\n${insert}\n---\n\n` + content;
|
||||
}
|
||||
|
||||
export function shouldNotifySync(quiet: boolean, written: number, errorCount: number): boolean {
|
||||
return !quiet || written > 0 || errorCount > 0;
|
||||
}
|
||||
|
||||
export class WatchlistController {
|
||||
private syncing = false;
|
||||
|
||||
constructor(private plugin: MediaDbPlugin) {}
|
||||
|
||||
private getKey(): string {
|
||||
|
|
@ -40,8 +46,8 @@ export class WatchlistController {
|
|||
const res = await obsidianFetch(new Request(url, { headers }));
|
||||
if (res.status === 429) {
|
||||
const err = new TmdbRateLimitError('TMDB 429');
|
||||
const ra = Number(res.headers.get('retry-after') ?? 2);
|
||||
err.retryAfterMs = ra * 1000;
|
||||
const ra = Number(res.headers.get('retry-after'));
|
||||
err.retryAfterMs = Number.isFinite(ra) && ra > 0 ? ra * 1000 : 2000;
|
||||
throw err;
|
||||
}
|
||||
if (res.status !== 200) throw new Error(`TMDB ${res.status} for ${url}`);
|
||||
|
|
@ -57,9 +63,12 @@ export class WatchlistController {
|
|||
listNotes: async () => {
|
||||
if (!(folder instanceof TFolder)) throw new Error(`Watchlist folder not found: ${this.plugin.settings.watchlistFolder}`);
|
||||
const files = folder.children.filter((f): f is TFile => f instanceof TFile && f.extension === 'md' && !f.name.startsWith('_'));
|
||||
const out: { path: string; content: string }[] = [];
|
||||
for (const f of files) out.push({ path: f.path, content: await app.vault.read(f) });
|
||||
return out;
|
||||
return files.map(f => ({ path: f.path }));
|
||||
},
|
||||
readNote: async path => {
|
||||
const f = app.vault.getAbstractFileByPath(path);
|
||||
if (!(f instanceof TFile)) throw new Error(`Watchlist note not found: ${path}`);
|
||||
return await app.vault.read(f);
|
||||
},
|
||||
writeNote: async (path, content) => {
|
||||
const f = app.vault.getAbstractFileByPath(path);
|
||||
|
|
@ -71,19 +80,30 @@ export class WatchlistController {
|
|||
};
|
||||
}
|
||||
|
||||
async syncNow(full: boolean, dryRun = false): Promise<SyncReport> {
|
||||
const key = this.getKey();
|
||||
const report = await syncFolder(this.makeDeps(key), { full, dryRun });
|
||||
const mode = dryRun ? 'DRY-RUN ' : '';
|
||||
new Notice(
|
||||
`Watchlist ${mode}sync: ${report.synced} checked, ${report.written} updated, ${report.flipped.length} flipped to Unwatched` +
|
||||
(report.errors.length ? `, ${report.errors.length} errors (see console)` : ''),
|
||||
);
|
||||
if (!dryRun) {
|
||||
this.plugin.settings.watchlistLastSync = Date.now();
|
||||
await this.plugin.saveSettings();
|
||||
async syncNow(full: boolean, dryRun = false, quiet = false): Promise<SyncReport> {
|
||||
if (this.syncing) {
|
||||
new Notice('Watchlist sync already running');
|
||||
return { scanned: 0, synced: 0, written: 0, skippedNoId: 0, skippedStatic: 0, flipped: [], errors: [] };
|
||||
}
|
||||
this.syncing = true;
|
||||
try {
|
||||
const key = this.getKey();
|
||||
const report = await syncFolder(this.makeDeps(key), { full, dryRun });
|
||||
const mode = dryRun ? 'DRY-RUN ' : '';
|
||||
if (shouldNotifySync(quiet, report.written, report.errors.length)) {
|
||||
new Notice(
|
||||
`Watchlist ${mode}sync: ${report.synced} checked, ${report.written} updated, ${report.flipped.length} flipped to Unwatched` +
|
||||
(report.errors.length ? `, ${report.errors.length} errors (see console)` : ''),
|
||||
);
|
||||
}
|
||||
if (!dryRun) {
|
||||
this.plugin.settings.watchlistLastSync = Date.now();
|
||||
await this.plugin.saveSettings();
|
||||
}
|
||||
return report;
|
||||
} finally {
|
||||
this.syncing = false;
|
||||
}
|
||||
return report;
|
||||
}
|
||||
|
||||
async maybeCatchUp(): Promise<void> {
|
||||
|
|
@ -91,47 +111,57 @@ export class WatchlistController {
|
|||
if (!s.watchlistEnabled) return;
|
||||
const due = s.watchlistLastSync + s.watchlistSyncIntervalHours * 3600_000;
|
||||
if (Date.now() >= due) {
|
||||
await this.syncNow(false).catch(e => console.error('[media-db-sync] scheduled sync failed', e));
|
||||
await this.syncNow(false, false, true).catch(e => console.error('[media-db-sync] scheduled sync failed', e));
|
||||
}
|
||||
}
|
||||
|
||||
async resolveMissingIds(dryRun = false): Promise<ResolveReport> {
|
||||
const key = this.getKey();
|
||||
const deps = this.makeDeps(key);
|
||||
const http = this.makeHttp();
|
||||
const search = (q: string, isMovie: boolean, year?: string) => searchTitle(http, key, q, isMovie, year);
|
||||
|
||||
const report: ResolveReport = { scanned: 0, resolved: 0, ambiguous: 0, errors: [] };
|
||||
const notes = await deps.listNotes();
|
||||
for (const note of notes) {
|
||||
const { frontmatter } = parseNote(note.content);
|
||||
if (noteTmdbRef(frontmatter)) continue;
|
||||
const type = stripQuotes(frontmatter['type']);
|
||||
if (type && type !== 'watchlist_item') continue; // skip type: list dashboards etc.
|
||||
report.scanned++;
|
||||
try {
|
||||
const filename = note.path.split('/').pop() ?? note.path;
|
||||
const result = await withRateLimitRetry(() => resolveNote(frontmatter, filename, search), deps.sleep);
|
||||
if (!result) {
|
||||
report.ambiguous++;
|
||||
deps.log(`ambiguous/no match: ${note.path}`);
|
||||
} else {
|
||||
report.resolved++;
|
||||
const mediaType = result.isMovie ? 'Movie' : 'TV Series';
|
||||
deps.log(`${dryRun ? '[dry] ' : ''}resolved ${note.path} -> tmdb_id ${result.tmdbId} (${result.matchedTitle})`);
|
||||
if (!dryRun) await deps.writeNote(note.path, patchFrontmatter(note.content, result.tmdbId, mediaType));
|
||||
}
|
||||
await deps.sleep(250);
|
||||
} catch (e) {
|
||||
report.errors.push({ path: note.path, error: e instanceof Error ? e.message : String(e) });
|
||||
deps.log(`ERROR ${note.path}: ${String(e)}`);
|
||||
}
|
||||
if (this.syncing) {
|
||||
new Notice('Watchlist sync already running');
|
||||
return { scanned: 0, resolved: 0, ambiguous: 0, errors: [] };
|
||||
}
|
||||
this.syncing = true;
|
||||
try {
|
||||
const key = this.getKey();
|
||||
const deps = this.makeDeps(key);
|
||||
const http = this.makeHttp();
|
||||
const search = (q: string, isMovie: boolean, year?: string) => searchTitle(http, key, q, isMovie, year);
|
||||
|
||||
const report: ResolveReport = { scanned: 0, resolved: 0, ambiguous: 0, errors: [] };
|
||||
const notes = await deps.listNotes();
|
||||
for (const note of notes) {
|
||||
const content = await deps.readNote(note.path);
|
||||
const { frontmatter } = parseNote(content);
|
||||
if (noteTmdbRef(frontmatter)) continue;
|
||||
const type = stripQuotes(frontmatter['type']);
|
||||
if (type && type !== 'watchlist_item') continue; // skip type: list dashboards etc.
|
||||
report.scanned++;
|
||||
try {
|
||||
const filename = note.path.split('/').pop() ?? note.path;
|
||||
const result = await withRateLimitRetry(() => resolveNote(frontmatter, filename, search), deps.sleep);
|
||||
if (!result) {
|
||||
report.ambiguous++;
|
||||
deps.log(`ambiguous/no match: ${note.path}`);
|
||||
} else {
|
||||
report.resolved++;
|
||||
const mediaType = result.isMovie ? 'Movie' : 'TV Series';
|
||||
deps.log(`${dryRun ? '[dry] ' : ''}resolved ${note.path} -> tmdb_id ${result.tmdbId} (${result.matchedTitle})`);
|
||||
if (!dryRun) await deps.writeNote(note.path, patchFrontmatter(content, result.tmdbId, mediaType));
|
||||
}
|
||||
await deps.sleep(250);
|
||||
} catch (e) {
|
||||
report.errors.push({ path: note.path, error: e instanceof Error ? e.message : String(e) });
|
||||
deps.log(`ERROR ${note.path}: ${String(e)}`);
|
||||
}
|
||||
}
|
||||
const mode = dryRun ? 'DRY-RUN ' : '';
|
||||
new Notice(
|
||||
`Watchlist ${mode}resolve: ${report.scanned} scanned, ${report.resolved} resolved, ${report.ambiguous} ambiguous/no match` +
|
||||
(report.errors.length ? `, ${report.errors.length} errors (see console)` : ''),
|
||||
);
|
||||
return report;
|
||||
} finally {
|
||||
this.syncing = false;
|
||||
}
|
||||
const mode = dryRun ? 'DRY-RUN ' : '';
|
||||
new Notice(
|
||||
`Watchlist ${mode}resolve: ${report.scanned} scanned, ${report.resolved} resolved, ${report.ambiguous} ambiguous/no match` +
|
||||
(report.errors.length ? `, ${report.errors.length} errors (see console)` : ''),
|
||||
);
|
||||
return report;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import { describe, expect, test } from 'bun:test';
|
||||
import { WatchlistController } from 'packages/obsidian/src/watchlist/WatchlistController';
|
||||
import { WatchlistController, shouldNotifySync } from 'packages/obsidian/src/watchlist/WatchlistController';
|
||||
import type { SyncDeps } from 'packages/obsidian/src/watchlist/SyncEngine';
|
||||
|
||||
function fakePlugin(overrides: Partial<{ enabled: boolean; last: number; hours: number }> = {}) {
|
||||
return {
|
||||
|
|
@ -38,3 +39,80 @@ describe('maybeCatchUp', () => {
|
|||
expect(called).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
function deferredDeps(): { deps: SyncDeps; listNotesCalls: () => number; release: () => void } {
|
||||
let listNotesCalls = 0;
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const deps: SyncDeps = {
|
||||
listNotes: async () => { listNotesCalls++; await gate; return []; },
|
||||
readNote: async () => '',
|
||||
writeNote: async () => {},
|
||||
fetchDetail: async () => ({}),
|
||||
sleep: async () => {},
|
||||
log: () => {},
|
||||
};
|
||||
return { deps, listNotesCalls: () => listNotesCalls, release };
|
||||
}
|
||||
|
||||
describe('concurrency guard', () => {
|
||||
test('overlapping syncNow calls: second short-circuits while first is in flight', async () => {
|
||||
const c = new WatchlistController(fakePlugin());
|
||||
const { deps, listNotesCalls, release } = deferredDeps();
|
||||
(c as any).getKey = () => 'fake-key';
|
||||
(c as any).makeDeps = () => deps;
|
||||
|
||||
const first = c.syncNow(false);
|
||||
const second = await c.syncNow(false);
|
||||
|
||||
expect(listNotesCalls()).toBe(1);
|
||||
expect(second.scanned).toBe(0);
|
||||
expect(second.written).toBe(0);
|
||||
|
||||
release();
|
||||
const firstResult = await first;
|
||||
expect(firstResult.scanned).toBe(0);
|
||||
});
|
||||
|
||||
test('syncNow in flight blocks resolveMissingIds (shared guard)', async () => {
|
||||
const c = new WatchlistController(fakePlugin());
|
||||
const { deps, release } = deferredDeps();
|
||||
(c as any).getKey = () => 'fake-key';
|
||||
(c as any).makeDeps = () => deps;
|
||||
|
||||
const first = c.syncNow(false);
|
||||
const resolveResult = await c.resolveMissingIds();
|
||||
|
||||
expect(resolveResult.scanned).toBe(0);
|
||||
expect(resolveResult.resolved).toBe(0);
|
||||
|
||||
release();
|
||||
await first;
|
||||
});
|
||||
|
||||
test('flag resets after completion → next call runs normally', async () => {
|
||||
const c = new WatchlistController(fakePlugin());
|
||||
(c as any).getKey = () => 'fake-key';
|
||||
let listNotesCalls = 0;
|
||||
(c as any).makeDeps = () => ({
|
||||
listNotes: async () => { listNotesCalls++; return []; },
|
||||
readNote: async () => '',
|
||||
writeNote: async () => {},
|
||||
fetchDetail: async () => ({}),
|
||||
sleep: async () => {},
|
||||
log: () => {},
|
||||
});
|
||||
|
||||
await c.syncNow(false);
|
||||
await c.syncNow(false);
|
||||
|
||||
expect(listNotesCalls).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe('shouldNotifySync', () => {
|
||||
test('non-quiet → always notifies', () => expect(shouldNotifySync(false, 0, 0)).toBe(true));
|
||||
test('quiet + no changes + no errors → suppressed', () => expect(shouldNotifySync(true, 0, 0)).toBe(false));
|
||||
test('quiet + written>0 → notifies', () => expect(shouldNotifySync(true, 3, 0)).toBe(true));
|
||||
test('quiet + errors>0 → notifies', () => expect(shouldNotifySync(true, 0, 2)).toBe(true));
|
||||
});
|
||||
|
|
|
|||
|
|
@ -23,16 +23,18 @@ keep me
|
|||
const AIRING_NOTE = ENDED_NOTE.replace('status: Ended', 'status: Returning Series').replace('last_air_date: 2023-11-09', 'last_air_date: 2023-10-01');
|
||||
|
||||
function makeDeps(notes: { path: string; content: string }[], detail: any = tvDetail) {
|
||||
const contents = new Map(notes.map(n => [n.path, n.content]));
|
||||
const writes: { path: string; content: string }[] = [];
|
||||
const fetches: string[] = [];
|
||||
const deps: SyncDeps = {
|
||||
listNotes: async () => notes,
|
||||
listNotes: async () => notes.map(n => ({ path: n.path })),
|
||||
readNote: async path => contents.get(path)!,
|
||||
writeNote: async (path, content) => { writes.push({ path, content }); },
|
||||
fetchDetail: async (id) => { fetches.push(id); return detail; },
|
||||
sleep: async () => {},
|
||||
log: () => {},
|
||||
};
|
||||
return { deps, writes, fetches };
|
||||
return { deps, writes, fetches, contents };
|
||||
}
|
||||
|
||||
describe('withRateLimitRetry', () => {
|
||||
|
|
@ -133,7 +135,8 @@ describe('syncFolder', () => {
|
|||
let calls = 0;
|
||||
const slept: number[] = [];
|
||||
const deps: SyncDeps = {
|
||||
listNotes: async () => [{ path: 'Loki.md', content: AIRING_NOTE }],
|
||||
listNotes: async () => [{ path: 'Loki.md' }],
|
||||
readNote: async () => AIRING_NOTE,
|
||||
writeNote: async () => {},
|
||||
fetchDetail: async () => {
|
||||
calls++;
|
||||
|
|
@ -160,4 +163,25 @@ describe('syncFolder', () => {
|
|||
expect(report.errors[0].path).toBe('Bad.md');
|
||||
expect(report.synced).toBe(1);
|
||||
});
|
||||
test('stale-read guard: content edited mid-sync is re-read fresh, not clobbered by early snapshot', async () => {
|
||||
const { deps, writes, contents } = makeDeps([
|
||||
{ path: 'A.md', content: AIRING_NOTE },
|
||||
{ path: 'B.md', content: AIRING_NOTE },
|
||||
]);
|
||||
const originalFetch = deps.fetchDetail;
|
||||
let calls = 0;
|
||||
deps.fetchDetail = async (id, isMovie) => {
|
||||
calls++;
|
||||
if (calls === 1) {
|
||||
// simulate the user editing B.md's "My Notes" while A.md is still mid-sync,
|
||||
// i.e. after listNotes() ran but before B.md is actually processed.
|
||||
contents.set('B.md', AIRING_NOTE.replace('keep me', 'edited during sync'));
|
||||
}
|
||||
return originalFetch(id, isMovie);
|
||||
};
|
||||
await syncFolder(deps);
|
||||
const bWrite = writes.find(w => w.path === 'B.md');
|
||||
expect(bWrite?.content).toContain('edited during sync');
|
||||
expect(bWrite?.content).not.toContain('keep me');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue