fix(fs): cap how many entry invalidations a node broadcasts per window

FSEntryStore now broadcasts at most 10,000 entries' cache invalidations
per broadcast flush window (outbound_flush_ms, 2 s by default). Past
that, keys are dropped locally only and peers fall back to the 60 s
entry-cache TTL; the store logs once per window when this happens.

Each flush goes to a peer as one POST that is re-sent whole on failure,
so a large recursive delete could otherwise build a batch the peer
rejects on every retry. The budget is per store instance, not per call,
so a delete issued page by page is bounded too, and single-entry writes
count against the same budget.

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<typeof FSEntryStore>[1],
+            server.stores as ConstructorParameters<typeof FSEntryStore>[2],
+        );
+
+    /** `count` files in a fresh directory, each cached under all its keys. */
+    const cachedFiles = async (count: number): Promise<FSEntry[]> => {
+        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<number> => {
+        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();

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<typeof puterStores>;
 
+    #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<void> {
-        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<void> {
@@ -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;
     }