diff --git a/src/backend/stores/fs/FSEntryStore.test.ts b/src/backend/stores/fs/FSEntryStore.test.ts index 2c5978e78..49814469f 100644 --- a/src/backend/stores/fs/FSEntryStore.test.ts +++ b/src/backend/stores/fs/FSEntryStore.test.ts @@ -27,7 +27,7 @@ import type { IConfig } from '../../types'; import { decodeCursor, encodeCursor } from '../../util/pagination.js'; import { generateDefaultFsentries } from '../../util/userProvisioning.js'; import type { FSEntry, FSEntryCreateInput } from './FSEntry.js'; -import { FSEntryStore } from './FSEntryStore.js'; +import { ENTRY_BROADCAST_CAP, FSEntryStore } from './FSEntryStore.js'; describe('FSEntryStore', () => { it('quotes camelCase storage allowance aliases for Postgres', async () => { @@ -447,6 +447,125 @@ describe('FSEntryStore cache invalidation', () => { }); }); +describe('FSEntryStore broadcast budget', () => { + // Its own store, so spending the budget can't starve the shared store's + // broadcasts that other tests assert on. + const budgetedStore = (outboundFlushMs?: number): FSEntryStore => + new FSEntryStore( + { + ...(configContainer as IConfig), + broadcast: { outbound_flush_ms: outboundFlushMs }, + }, + server.clients as ConstructorParameters[1], + server.stores as ConstructorParameters[2], + ); + + /** `count` files in a fresh directory, each cached under all its keys. */ + const cachedFiles = async (count: number): Promise => { + const user = await makeUser(); + const dir = `${user.home}/bulk-${uuidv4()}`; + const entries = await store.batchCreateEntries( + Array.from({ length: count }, (_, i) => ({ + userId: user.userId, + uuid: uuidv4(), + path: `${dir}/f${i}`, + size: 1, + bucket: 'puter-local', + bucketRegion: 'us-west-2', + })), + ); + await store.getEntriesByIds(entries.map((entry) => entry.id)); + return entries; + }; + + const cachedKeyCount = async (entries: FSEntry[]): Promise => { + const values = await server.clients.redis.mget( + entries.flatMap(entryCacheKeys), + ); + return values.filter((value) => value !== null).length; + }; + + const broadcastEntryCount = (updates: CacheUpdate[]): number => + updates + .flatMap((update) => update.cacheKey) + .filter((key) => key.startsWith('prodfsv2:fsentry:uuid:')).length; + + it('broadcasts at most the cap in one window and invalidates the rest locally', async () => { + const entries = await cachedFiles(ENTRY_BROADCAST_CAP + 50); + expect(await cachedKeyCount(entries)).toBe(entries.length * 3); + const budgeted = budgetedStore(); + + vi.useFakeTimers({ toFake: ['Date'] }); + try { + const updates = await captureCacheUpdates(() => + budgeted.deleteEntries(entries), + ); + expect(broadcastEntryCount(updates)).toBe(ENTRY_BROADCAST_CAP); + } finally { + vi.useRealTimers(); + } + expect(await cachedKeyCount(entries)).toBe(0); + }); + + it('caps a delete that arrives page by page, warning once', async () => { + const entries = await cachedFiles(ENTRY_BROADCAST_CAP + 50); + const budgeted = budgetedStore(); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + vi.useFakeTimers({ toFake: ['Date'] }); + try { + const updates = await captureCacheUpdates(async () => { + for (let i = 0; i < entries.length; i += 150) { + await budgeted.deleteEntries(entries.slice(i, i + 150)); + } + }); + expect(broadcastEntryCount(updates)).toBe(ENTRY_BROADCAST_CAP); + const capWarnings = warn.mock.calls.filter(([message]) => + String(message).startsWith('[FSEntryStore]'), + ); + expect(capWarnings).toHaveLength(1); + } finally { + vi.useRealTimers(); + warn.mockRestore(); + } + expect(await cachedKeyCount(entries)).toBe(0); + }); + + it('starts a fresh budget once the configured flush window passes', async () => { + const entries = await cachedFiles(ENTRY_BROADCAST_CAP + 6); + const spend = entries.slice(0, ENTRY_BROADCAST_CAP); + const late = entries.slice(ENTRY_BROADCAST_CAP, -3); + const next = entries.slice(-3); + const budgeted = budgetedStore(5_000); + + vi.useFakeTimers({ toFake: ['Date'] }); + try { + const start = Date.now(); + const spent = await captureCacheUpdates(() => + budgeted.deleteEntries(spend), + ); + expect(broadcastEntryCount(spent)).toBe(ENTRY_BROADCAST_CAP); + + vi.setSystemTime(start + 4_999); + const capped = await captureCacheUpdates(() => + budgeted.deleteEntries(late), + ); + expect(broadcastEntryCount(capped)).toBe(0); + expect(await cachedKeyCount(late)).toBe(0); + + vi.setSystemTime(start + 5_000); + const fresh = await captureCacheUpdates(() => + budgeted.deleteEntries(next), + ); + expect(fresh.flatMap((update) => update.cacheKey)).toEqual( + expect.arrayContaining(next.flatMap(entryCacheKeys)), + ); + } finally { + vi.useRealTimers(); + } + }); +}); + describe('FSEntryStore legacy NULL-path healing', () => { it('resolves and backfills a path-less row looked up by path', async () => { const user = await makeUser(); diff --git a/src/backend/stores/fs/FSEntryStore.ts b/src/backend/stores/fs/FSEntryStore.ts index 3a1736ca2..212ee0ae3 100644 --- a/src/backend/stores/fs/FSEntryStore.ts +++ b/src/backend/stores/fs/FSEntryStore.ts @@ -103,6 +103,13 @@ const vacatedHomeKey = (username: string) => `prodfsv2:vacated-home:${username.toLowerCase()}`; const BULK_QUERY_CHUNK_SIZE = 200; const DEFAULT_DB_CHUNK_CONCURRENCY = 4; +// Most entries whose invalidations one node broadcasts per broadcast flush +// window. A flush reaches each peer as one POST (50 MB max, ~200 bytes an +// entry); past the cap, invalidations stay local and peers wait out +// ENTRY_CACHE_TTL_SECONDS. +export const ENTRY_BROADCAST_CAP = 10_000; +// BroadcastService's default `broadcast.outbound_flush_ms`. +const DEFAULT_BROADCAST_FLUSH_MS = 2_000; /** * Store backing the `fsentries` table. Owns DB CRUD over filesystem entries, @@ -112,6 +119,10 @@ const DEFAULT_DB_CHUNK_CONCURRENCY = 4; export class FSEntryStore extends PuterStore { declare protected stores: LayerInstances; + #broadcastWindowStart = 0; + #broadcastedInWindow = 0; + #broadcastCapLogged = false; + #insertIgnoreIntoFsentriesSql(): string { return this.clients.db.insertIgnoreInto('fsentries'); } @@ -353,12 +364,48 @@ export class FSEntryStore extends PuterStore { } /** - * Drops here and in peer regions, as one event for the whole batch. The - * read-path `#writeEntryToCache` stays local so backfills don't fan out. + * Drops here and, within the window's broadcast budget, in peer regions as + * one event for the batch. The read-path `#writeEntryToCache` stays local + * so backfills don't fan out. */ async #invalidateEntriesCache(entries: FSEntry[]): Promise { - const keys = entries.flatMap((entry) => this.#entryCacheKeys(entry)); - await this.publishCacheKeys({ keys, broadcast: true }); + const keysOf = (batch: FSEntry[]) => + batch.flatMap((entry) => this.#entryCacheKeys(entry)); + const broadcastCount = this.#takeBroadcastBudget(entries.length); + await this.publishCacheKeys({ + keys: keysOf(entries.slice(0, broadcastCount)), + broadcast: true, + }); + await this.publishCacheKeys({ + keys: keysOf(entries.slice(broadcastCount)), + }); + } + + /** How many of `count` entries may still broadcast this window. */ + #takeBroadcastBudget(count: number): number { + const flushMs = Number(this.config.broadcast?.outbound_flush_ms); + const windowMs = + Number.isFinite(flushMs) && flushMs > 0 + ? flushMs + : DEFAULT_BROADCAST_FLUSH_MS; + const now = Date.now(); + if (now - this.#broadcastWindowStart >= windowMs) { + this.#broadcastWindowStart = now; + this.#broadcastedInWindow = 0; + this.#broadcastCapLogged = false; + } + const allowed = Math.min( + count, + ENTRY_BROADCAST_CAP - this.#broadcastedInWindow, + ); + this.#broadcastedInWindow += allowed; + if (allowed < count && !this.#broadcastCapLogged) { + this.#broadcastCapLogged = true; + console.warn( + `[FSEntryStore] broadcast cap (${ENTRY_BROADCAST_CAP} entries per ${windowMs} ms) reached; invalidating the rest locally only`, + ); + } + return allowed; } async invalidateEntryCacheByPathForUser( @@ -382,7 +429,10 @@ export class FSEntryStore extends PuterStore { return; } - await this.publishCacheKeys({ keys: cacheKeys, broadcast: true }); + await this.publishCacheKeys({ + keys: cacheKeys, + broadcast: this.#takeBroadcastBudget(1) > 0, + }); } async invalidateEntryCacheByUuid(uuid: string): Promise { @@ -412,7 +462,7 @@ export class FSEntryStore extends PuterStore { await this.publishCacheKeys({ keys: [`prodfsv2:fsentry:uuid:${uuid}`], - broadcast: true, + broadcast: this.#takeBroadcastBudget(1) > 0, }); } @@ -443,7 +493,7 @@ export class FSEntryStore extends PuterStore { await this.publishCacheKeys({ keys: [`prodfsv2:fsentry:id:${id}`], - broadcast: true, + broadcast: this.#takeBroadcastBudget(1) > 0, }); } @@ -1306,7 +1356,7 @@ export class FSEntryStore extends PuterStore { keys: this.#entryCacheKeys(entry), serializedData: JSON.stringify(entry), ttlSeconds: ENTRY_CACHE_TTL_SECONDS, - broadcast: true, + broadcast: this.#takeBroadcastBudget(1) > 0, }); return entry; }