diff --git a/packages/realm-server/lib/realm-file-changes-listener.ts b/packages/realm-server/lib/realm-file-changes-listener.ts index 56e919107da..152530ab9d0 100644 --- a/packages/realm-server/lib/realm-file-changes-listener.ts +++ b/packages/realm-server/lib/realm-file-changes-listener.ts @@ -13,14 +13,21 @@ const log = logger('realm-server:file-changes-listener'); // runtime-common/realm.ts), every listener subscribed on this channel looks // up the URL in its lookup function. If the realm is mounted locally, // `realm.invalidateCache(path)` clears the matching #sourceCache / -// #transpiledModuleCache entries. If it's not mounted, the notification is dropped — -// this instance has no stale state to clear. +// #transpiledModuleCache entries, and `realm.refreshDirectoryView(path)` +// re-lists the path's parent directory so the kernel's cached view of the +// shared filesystem (which can still say the file is absent) is refreshed +// too. If it's not mounted, the notification is dropped — this instance has +// no stale state to clear. // // Bulk variant: when the path is the wildcard sentinel `*` (CS-11156), // `realm.clearLocalSourceCaches()` drops every cached path for that realm. Emitted // by the publish-realm / unpublish-realm / delete-realm handlers after the // FS swap or removal so peers (whose file-watcher events do NOT cross -// replicas) bypass their pre-swap cached bytes on the next read. +// replicas) bypass their pre-swap cached bytes on the next read. Note the +// wildcard branch clears only the in-memory byte caches: the kernel's view of +// the swapped directory tree is NOT refreshed here (that would take a +// recursive walk of the realm), so a peer can still serve stale +// file-not-found for files a republish added, until the kernel cache expires. // // The LISTEN is backed by `PgAdapter.subscribe` (shared multiplexed // notification client). There is no periodic work to run between @@ -102,6 +109,13 @@ export class RealmFileChangesListener { realm.clearLocalSourceCaches(); } else { realm.invalidateCache(parsed.path); + // Fire-and-forget: the NOTIFY handler stays synchronous, and a failed + // directory listing only means the kernel cache expires on its own. + realm.refreshDirectoryView(parsed.path).catch((err: unknown) => { + log.warn( + `refreshDirectoryView failed for ${parsed.url} ${parsed.path}: ${String(err)}`, + ); + }); } } catch (err: unknown) { const op = isWildcard ? 'clearLocalSourceCaches' : 'invalidateCache'; diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 1e398da767d..a2c386f00c6 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -18,10 +18,9 @@ import type { LocalPath } from '@cardstack/runtime-common/paths'; import type { ServerResponse } from 'http'; import sane, { type Watcher } from 'sane'; -import type { ReadStream, Stats } from 'fs-extra'; +import type { Dirent, ReadStream, Stats } from 'fs-extra'; import fsExtra from 'fs-extra'; const { - readdirSync, existsSync, writeFileSync, statSync, @@ -61,26 +60,6 @@ function statIfExists(absolutePath: string): Stats | undefined { } } -// The same vanishing-path race applies to a directory listing: a concurrent -// delete — e.g. a published realm being unpublished while a mtimes traversal is -// descending its tree — can remove a directory between a parent listing that -// yielded it and the recursive read that opens it, so treat a vanished -// directory as empty rather than letting the raw ENOENT escape. Only ENOENT is -// swallowed: unlike statIfExists (whose callers probe arbitrary paths), the -// traversal only ever reads directories it just listed, and ENOTDIR — the -// target is a regular file — is a genuine not-a-directory that callers such as -// directoryEntries must still see. -function readdirIfExists(absolutePath: string) { - try { - return readdirSync(absolutePath, { withFileTypes: true }); - } catch (err: any) { - if (err?.code === 'ENOENT') { - return undefined; - } - throw err; - } -} - function parseMatrixSendEventError(error: unknown): { status?: number; errcode?: string; @@ -140,11 +119,20 @@ export class NodeAdapter implements RealmAdapter { ensureDirSync(path); } let absolutePath = join(this.realmDir, path); - let entries = readdirIfExists(absolutePath); - if (!entries) { - return; + // Asynchronous on purpose: on a network filesystem a listing is a round + // trip, and this generator runs on the request path and in the + // cross-instance directory refresh, neither of which may block the event + // loop. A missing directory lists as empty. + let entries: Dirent[]; + try { + entries = await fsExtra.readdir(absolutePath, { withFileTypes: true }); + } catch (err: any) { + if (err?.code === 'ENOENT') { + return; + } + throw err; } - for await (let entry of entries) { + for (let entry of entries) { let isDirectory = entry.isDirectory(); let isFile = entry.isFile(); if (!isDirectory && !isFile) { diff --git a/packages/realm-server/tests/directory-view-refresher-test.ts b/packages/realm-server/tests/directory-view-refresher-test.ts new file mode 100644 index 00000000000..8b476a136bc --- /dev/null +++ b/packages/realm-server/tests/directory-view-refresher-test.ts @@ -0,0 +1,162 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import { basename } from 'path'; +import { + DirectoryViewRefresher, + ancestorDirectories, +} from '@cardstack/runtime-common'; + +module(basename(import.meta.filename), function () { + test('ancestorDirectories lists the realm root first, then each directory down to the parent', function (assert) { + assert.deepEqual(ancestorDirectories('c.json'), ['']); + assert.deepEqual(ancestorDirectories('a/c.json'), ['', 'a']); + assert.deepEqual(ancestorDirectories('a/b/c.json'), ['', 'a', 'a/b']); + }); + + test('refresh lists every ancestor, root first', async function (assert) { + let listed: string[] = []; + let refresher = new DirectoryViewRefresher(async (dir) => { + listed.push(dir); + }); + + await refresher.refresh('PersonCard/nested/bob.json'); + + assert.deepEqual(listed, ['', 'PersonCard', 'PersonCard/nested']); + }); + + test('concurrent refreshes of one directory never overlap listings and coalesce into fewer listings than refreshes', async function (assert) { + let calls: string[] = []; + let inFlight = new Map(); + let maxInFlight = new Map(); + let refresher = new DirectoryViewRefresher(async (dir) => { + calls.push(dir); + let now = (inFlight.get(dir) ?? 0) + 1; + inFlight.set(dir, now); + maxInFlight.set(dir, Math.max(maxInFlight.get(dir) ?? 0, now)); + // A listing takes a round trip; every refresh below arrives during it. + await new Promise((r) => setTimeout(r, 5)); + inFlight.set(dir, now - 1); + }); + + // Five notifications for files in the same directory arrive at once (a + // batch write emits one per file). + let names = ['a.json', 'b.json', 'c.json', 'd.json', 'e.json']; + await Promise.all( + names.map((name) => refresher.refresh(`PersonCard/${name}`)), + ); + + let count = (dir: string) => calls.filter((c) => c === dir).length; + assert.strictEqual(maxInFlight.get(''), 1, 'root listings never overlap'); + assert.strictEqual( + maxInFlight.get('PersonCard'), + 1, + 'PersonCard listings never overlap', + ); + // The five refresh() calls reach their first `await` synchronously, so + // the first starts the root listing and the other four share one pending + // follow-up: the root count is structurally exactly 2. The PersonCard + // count depends on how its first listing interleaves with the root + // follow-up, so a bound is the honest assertion there. + assert.strictEqual( + count(''), + 2, + 'root listed once, then one follow-up for the four that arrived during it', + ); + let dirBounded = count('PersonCard') <= 3; + assert.true( + dirBounded, + `PersonCard listings stayed bounded (${count('PersonCard')})`, + ); + + calls.length = 0; + await refresher.refresh('PersonCard/f.json'); + assert.deepEqual( + calls, + ['', 'PersonCard'], + 'after everything settles a new refresh lists again', + ); + }); + + test('a request that arrives while the follow-up listing is running queues another one', async function (assert) { + let resolvers: Array<() => void> = []; + let refresher = new DirectoryViewRefresher( + (_dir) => + new Promise((resolve) => { + resolvers.push(resolve); + }), + ); + let started = () => resolvers.length; + let untilStarted = async (n: number) => { + while (started() < n) { + await new Promise((r) => setTimeout(r, 0)); + } + }; + + let first = refresher.refresh('a.json'); // starts listing 1 + let second = refresher.refresh('b.json'); // queues listing 2 + assert.strictEqual(started(), 1, 'only listing 1 has started'); + + resolvers[0](); + await untilStarted(2); // listing 2 is now running + + // This request arrives while listing 2 runs; listing 2 started before the + // write behind this request, so it must get listing 3, not share 2. + let third = refresher.refresh('c.json'); + resolvers[1](); + await untilStarted(3); + resolvers[2](); + + await Promise.all([first, second, third]); + assert.strictEqual(started(), 3, 'the late request got its own listing'); + }); + + test('one ancestor failing does not stop the rest of the chain', async function (assert) { + let listed: string[] = []; + let refresher = new DirectoryViewRefresher(async (dir) => { + listed.push(dir); + if (dir === '') { + throw new Error('root EIO'); + } + }); + + await assert.rejects(refresher.refresh('a/b/c.json'), /root EIO/); + assert.deepEqual( + listed, + ['', 'a', 'a/b'], + 'the parent directories were still listed after the root failed', + ); + }); + + test('a listing that never settles is timed out and stops blocking later refreshes', async function (assert) { + let calls = 0; + let refresher = new DirectoryViewRefresher( + (_dir) => + new Promise((resolve) => { + calls++; + if (calls > 1) { + resolve(); + } + // the first listing never settles + }), + { timeoutMs: 20 }, + ); + + await assert.rejects(refresher.refresh('a.json'), /timed out listing ""/); + await refresher.refresh('b.json'); + assert.strictEqual(calls, 2, 'the second refresh got a fresh listing'); + }); + + test('a failed listing rejects that refresh and does not poison later ones', async function (assert) { + let attempt = 0; + let refresher = new DirectoryViewRefresher(async () => { + attempt++; + if (attempt === 1) { + throw new Error('EIO'); + } + }); + + await assert.rejects(refresher.refresh('x.json'), /EIO/); + await refresher.refresh('x.json'); + assert.strictEqual(attempt, 2, 'the second refresh listed again'); + }); +}); diff --git a/packages/realm-server/tests/index.ts b/packages/realm-server/tests/index.ts index cd2e2783054..bfee94f94de 100644 --- a/packages/realm-server/tests/index.ts +++ b/packages/realm-server/tests/index.ts @@ -310,6 +310,7 @@ const ALL_TEST_FILES: string[] = [ './realm-registry-reconciler-test', './realm-registry-writes-test', './realm-file-changes-listener-test', + './directory-view-refresher-test', './realm-index-updated-listener-test', './jobs-finished-listener-test', './realm-routing-test', diff --git a/packages/realm-server/tests/realm-file-changes-listener-test.ts b/packages/realm-server/tests/realm-file-changes-listener-test.ts index 8bc52a81288..56425cef220 100644 --- a/packages/realm-server/tests/realm-file-changes-listener-test.ts +++ b/packages/realm-server/tests/realm-file-changes-listener-test.ts @@ -10,13 +10,14 @@ import { } from '../lib/realm-file-changes-listener.ts'; // Minimal fake `Realm` — the listener calls `.url` (via lookup), -// `.invalidateCache(path)` for per-path payloads, and `.clearLocalSourceCaches()` -// for wildcard payloads (CS-11156). Stub both; tests pick whichever they -// care about. +// `.invalidateCache(path)` plus `.refreshDirectoryView(path)` for per-path +// payloads, and `.clearLocalSourceCaches()` for wildcard payloads. +// Stub all three; tests pick whichever they care about. function makeFakeRealm( url: string, hooks: { onInvalidate?: (path: string) => void; + onRefreshDirectory?: (path: string) => void; onClearAll?: () => void; }, ): Realm { @@ -25,6 +26,9 @@ function makeFakeRealm( invalidateCache(path: string) { hooks.onInvalidate?.(path); }, + async refreshDirectoryView(path: string) { + hooks.onRefreshDirectory?.(path); + }, clearLocalSourceCaches() { hooks.onClearAll?.(); }, @@ -103,9 +107,11 @@ module(basename(import.meta.filename), function () { module('RealmFileChangesListener (dispatch)', function () { test('handleNotification forwards to the mounted realm', function (assert) { const invalidations: Array<{ url: string; path: string }> = []; + const refreshedDirectories: string[] = []; const realmA = makeFakeRealm('http://x.test/a/', { onInvalidate: (path) => invalidations.push({ url: 'http://x.test/a/', path }), + onRefreshDirectory: (path) => refreshedDirectories.push(path), }); const listener = new RealmFileChangesListener({ dbAdapter: {} as unknown as PgAdapter, @@ -118,13 +124,45 @@ module(basename(import.meta.filename), function () { assert.deepEqual(invalidations, [ { url: 'http://x.test/a/', path: 'cards/foo.gts' }, ]); + assert.deepEqual( + refreshedDirectories, + ['cards/foo.gts'], + 'the peer re-lists the directory holding the written path', + ); + }); + + test('handleNotification keeps going when refreshDirectoryView rejects', async function (assert) { + const invalidations: string[] = []; + const realmA = { + url: 'http://x.test/a/', + invalidateCache(path: string) { + invalidations.push(path); + }, + async refreshDirectoryView() { + throw new Error('EIO'); + }, + clearLocalSourceCaches() {}, + } as unknown as Realm; + const listener = new RealmFileChangesListener({ + dbAdapter: {} as unknown as PgAdapter, + lookupMountedRealm: (url) => + url === 'http://x.test/a/' ? realmA : undefined, + }); + + listener.handleNotification('http://x.test/a/:cards/foo.gts'); + // let the rejected promise settle; an unhandled rejection would fail the run + await new Promise((resolve) => setTimeout(resolve, 0)); + + assert.deepEqual(invalidations, ['cards/foo.gts']); }); test('handleNotification with wildcard payload calls clearLocalSourceCaches and not invalidateCache (CS-11156)', function (assert) { const invalidations: string[] = []; + const refreshedDirectories: string[] = []; const clearAllCount = { value: 0 }; const realmA = makeFakeRealm('http://x.test/a/', { onInvalidate: (path) => invalidations.push(path), + onRefreshDirectory: (path) => refreshedDirectories.push(path), onClearAll: () => clearAllCount.value++, }); const listener = new RealmFileChangesListener({ @@ -145,6 +183,11 @@ module(basename(import.meta.filename), function () { [], 'invalidateCache not called for wildcard payload', ); + assert.deepEqual( + refreshedDirectories, + [], + 'refreshDirectoryView not called for wildcard payload', + ); }); test('handleNotification drops silently when the url is not mounted', function (assert) { diff --git a/packages/runtime-common/directory-view-refresher.ts b/packages/runtime-common/directory-view-refresher.ts new file mode 100644 index 00000000000..5e29ad81baf --- /dev/null +++ b/packages/runtime-common/directory-view-refresher.ts @@ -0,0 +1,148 @@ +import type { LocalPath } from './paths.ts'; + +// Refreshes a realm-server instance's view of the directories that hold a +// path another instance just wrote or deleted. +// +// The realm directory is a shared network filesystem (EFS/NFS) when several +// realm-server instances run at once, and each instance's kernel caches its +// own view of every directory — including "this name is not here" answers +// from earlier lookups. A file a peer just wrote therefore keeps reading as +// missing on this instance until the directory's attribute cache expires, +// tens of seconds later. Listing a directory makes the kernel re-read it from +// the server, which discards the stale negative entries, so the next lookup of +// the path sees the file. The listing is done for every ancestor, root first, +// because the peer's write may have created the parent directories too and a +// stale "not here" for `new-dir` would otherwise hide `new-dir/file` even +// after `new-dir` itself is listed. +// +// Notifications arrive one per written path, so a batch write into one +// directory would list that directory once per file. Listings are coalesced +// per directory: while one is running, later requests wait for one pending +// follow-up listing instead of starting their own (a request that arrives +// after a listing started cannot know whether that listing already saw its +// write, so it needs one that starts later). When the pending listing starts +// it becomes the running one, so a request arriving during it can queue a +// fresh follow-up in turn. +// +// Budget: a single-path notification costs one listing per ancestor (realm +// root included) on every instance with the realm mounted — the emitting +// instance receives its own NOTIFY too. That is deliberate: realm trees are +// shallow, bursts coalesce per directory, and the simpler shape is easier to +// reason about than walking up only when the written name is missing from its +// parent's entries. +// +// Containment: one ancestor failing does not stop the rest of the chain — +// the immediate parent is the listing that matters, and the realm root (first +// in the chain, always present) is the one least worth dying for. A listing +// that never settles is timed out so it stops occupying the per-directory +// slot; the kernel's own cache TTL remains the backstop either way. +interface DirectoryListing { + running: Promise; + // Requested but not yet started; becomes `running` when it starts. + pending?: Promise; +} + +export class DirectoryViewRefresher { + #listings = new Map(); + + #listDirectory: (directory: LocalPath) => Promise; + #timeoutMs: number; + + constructor( + listDirectory: (directory: LocalPath) => Promise, + opts?: { timeoutMs?: number }, + ) { + this.#listDirectory = listDirectory; + this.#timeoutMs = opts?.timeoutMs ?? 15_000; + } + + async refresh(path: LocalPath): Promise { + let errors: unknown[] = []; + for (let directory of ancestorDirectories(path)) { + try { + await this.#list(directory); + } catch (err) { + errors.push(err); + } + } + if (errors.length === 1) { + throw errors[0]; + } + if (errors.length > 1) { + throw new AggregateError(errors, 'directory refresh partially failed'); + } + } + + #list(directory: LocalPath): Promise { + let listing = this.#listings.get(directory); + if (!listing) { + return this.#startListing(directory); + } + if (!listing.pending) { + listing.pending = listing.running + .then( + () => undefined, + () => undefined, + ) + .then(() => this.#startListing(directory)); + } + // The pending listing has not started yet, so it is guaranteed to observe + // the write behind this request. + return listing.pending; + } + + #startListing(directory: LocalPath): Promise { + // The timeout does not cancel the underlying listing — it releases the + // per-directory slot, so a hung filesystem call stops blocking every + // later refresh of this directory for the life of the process. A fresh + // listing may then overlap the hung one; overlapping reads are safe. + let running = this.#withTimeout(this.#listDirectory(directory), directory); + let listing: DirectoryListing = { running }; + // Replaces any predecessor (whose `pending` was this very listing), so a + // request arriving from here on queues behind this listing instead of + // sharing it. + this.#listings.set(directory, listing); + let cleanup = () => { + if (this.#listings.get(directory) === listing && !listing.pending) { + this.#listings.delete(directory); + } + }; + running.then(cleanup, cleanup); + return running; + } + + #withTimeout(listing: Promise, directory: LocalPath): Promise { + return new Promise((resolve, reject) => { + let timer = setTimeout(() => { + reject( + new Error( + `timed out listing "${directory}" after ${this.#timeoutMs}ms`, + ), + ); + }, this.#timeoutMs); + // In Node, don't let a pending timer hold the process open. + (timer as { unref?: () => void }).unref?.(); + listing.then( + () => { + clearTimeout(timer); + resolve(); + }, + (err) => { + clearTimeout(timer); + reject(err); + }, + ); + }); + } +} + +// The directories that lead to `path`, realm root first: for `a/b/c.json` +// this is `''`, `a`, `a/b`. +export function ancestorDirectories(path: LocalPath): LocalPath[] { + let directories: LocalPath[] = ['']; + let segments = path.split('/'); + for (let i = 1; i < segments.length; i++) { + directories.push(segments.slice(0, i).join('/')); + } + return directories; +} diff --git a/packages/runtime-common/index.ts b/packages/runtime-common/index.ts index dc2ad8b3cac..4bc4046e476 100644 --- a/packages/runtime-common/index.ts +++ b/packages/runtime-common/index.ts @@ -1114,6 +1114,7 @@ import { Loader } from './loader.ts'; export * from './frontmatter-parse.ts'; export * from './http-range.ts'; export * from './paths.ts'; +export * from './directory-view-refresher.ts'; export * from './realm-client.ts'; export * from './realm-operations.ts'; export * from './published-realm-url.ts'; diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index 2452dbd434c..024190dd720 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -206,6 +206,7 @@ import type { import { RealmAuthDataSource } from './realm-auth-data-source.ts'; import { AliasCache } from './cache/alias-cache.ts'; +import { DirectoryViewRefresher } from './directory-view-refresher.ts'; import { fetcher } from './fetcher.ts'; import { RealmIndexQueryEngine } from './realm-index-query-engine.ts'; import { RealmIndexUpdater } from './realm-index-updater.ts'; @@ -905,6 +906,11 @@ export class Realm { #definitionLookup: DefinitionLookup; #copiedFromRealm: URL | undefined; #sourceCache = new AliasCache(); + #directoryViewRefresher = new DirectoryViewRefresher(async (directory) => { + for await (let _entry of this.#adapter.readdir(directory)) { + // draining the listing is the whole effect; the entries are not used + } + }); // Per-path generation counters for #sourceCache — the source-read analogue // of #transpiledModuleCacheGenerations below. getSourceOrRedirect reads // bytes from disk under an `await` (getFileWithFallbacks + materializeFileRef) @@ -1980,6 +1986,14 @@ export class Realm { } } + // Refresh this instance's filesystem view of the directories that hold + // `path`, after a peer instance wrote or deleted that path. See + // DirectoryViewRefresher for why a shared-filesystem peer needs this and how + // repeated requests for one directory are coalesced. + refreshDirectoryView(path: LocalPath): Promise { + return this.#directoryViewRefresher.refresh(path); + } + // CS-11028: shared drop helper for any in-process site that invalidates a // single #transpiledModuleCache entry — writeMany, delete/deleteAll, the local // file-watcher callback, the index-updater's executable-invalidation