From d88076c344a2cb98d9963712c64e1e7067ded6b7 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Thu, 8 Oct 2026 20:00:56 -0700 Subject: [PATCH] fix(events): forward kv values only to regions with rows that ask - A region counts its session rows asking for kv values per token and announces the token's value variant (`v#`) over the existing `watch` item on the first such row, withdrawing it with the last. Add, remove, reap, reanchor and refresh all keep it, reading `includeValue` off the row as it is dropped. - A kv write reads the value variants alongside the watch index and attaches the value only to forwards bound for a region that announced one. --- .../services/events/EventsService.test.ts | 37 +++ src/backend/services/events/EventsService.ts | 68 ++++-- .../services/events/forwarding.test.ts | 126 +++++++++- .../events/EventSubscriptionStore.test.ts | 104 +++++++++ .../stores/events/EventSubscriptionStore.ts | 215 +++++++++++++----- 5 files changed, 469 insertions(+), 81 deletions(-) diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index 7ee4cd7d2..18e27d4be 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -33,6 +33,7 @@ import type { Actor } from '../../core/actor.js'; import { isHttpError } from '../../core/http/HttpError.js'; import { EventSubscriptionStore, + valueWatchToken, type DurableSubscription, } from '../../stores/events/EventSubscriptionStore.js'; import type { KvShareHandle } from '../../stores/events/KvShareHandleStore.js'; @@ -2809,6 +2810,42 @@ describe('the cross-app kv gate', () => { expect(sent).toEqual([]); }); + it('sends no value to a peer whose rows did not ask for one', async () => { + await store.noteRemoteWatch( + userId, + kvAnchorToken(`user-${userId}`, OWN_APP, 'cart'), + 'east', + 'add', + ); + + await dispatchKv(['cart'], { values: [{ items: 3 }] }); + + expect(forwarded).toHaveLength(1); + expect(forwarded[0].regions).toEqual(['east']); + expect(forwarded[0].item.kv).not.toHaveProperty('value'); + }); + + it('sends the value only to the peers whose rows asked for it', async () => { + const token = kvAnchorToken(`user-${userId}`, OWN_APP, 'cart'); + for (const region of ['east', 'south']) + await store.noteRemoteWatch(userId, token, region, 'add'); + await store.noteRemoteWatch( + userId, + valueWatchToken(token), + 'east', + 'add', + ); + + await dispatchKv(['cart'], { values: [{ items: 3 }] }); + + const to = (region: string) => + forwarded.find((call) => call.regions.includes(region)); + expect(forwarded).toHaveLength(2); + expect(to('east')?.regions).toEqual(['east']); + expect(to('east')?.item.kv).toMatchObject({ value: { items: 3 } }); + expect(to('south')?.item.kv).not.toHaveProperty('value'); + }); + it('marks a private key on the way to a peer region', async () => { vi.useFakeTimers(); // A remote watcher is what makes `dispatchKv` forward at all. diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index a9d076229..9eed91715 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -3817,34 +3817,62 @@ export class EventsService extends PuterService { }; const tokensPerKey = contexts.map((context) => subject.tokens(context)); - const { local, remote } = - await this.stores.eventSubscription.watchedFor(ownerUserId, [ - ...new Set(tokensPerKey.flat()), - ]); + const { local, remote, remoteValues } = + await this.stores.eventSubscription.watchedFor( + ownerUserId, + [...new Set(tokensPerKey.flat())], + { values: true }, + ); if (local.length === 0 && remote.size === 0) return false; if (!options.forwarded && remote.size > 0) contexts.forEach((context, i) => { const regions = new Set(); - for (const token of tokensPerKey[i]) + const asking = new Set(); + for (const token of tokensPerKey[i]) { for (const region of remote.get(token) ?? []) regions.add(region); + for (const region of remoteValues.get(token) ?? []) + asking.add(region); + } if (regions.size === 0) return; - this.services.eventForward.forwardEvent([...regions], { - family: 'kv', - ownerUserId, - actingUserId: options.actingUserId, - id: context.id, - ts: context.ts, - kv: { - userUuid: namespace.userUuid, - appUid: namespace.appUid, - kvKey: context.kvKey, - op: context.op, - ...valueAt(i), - ...(context.noShare ? { noShare: true as const } : {}), - }, - }); + // The value goes only where a row asked for it; everywhere + // else it would cross regions to be discarded on arrival. + const value = [...regions].some((region) => asking.has(region)) + ? valueAt(i) + : undefined; + const forward = ( + to: string[], + carried?: { value: unknown }, + ) => { + if (to.length === 0) return; + this.services.eventForward.forwardEvent(to, { + family: 'kv', + ownerUserId, + actingUserId: options.actingUserId, + id: context.id, + ts: context.ts, + kv: { + userUuid: namespace.userUuid, + appUid: namespace.appUid, + kvKey: context.kvKey, + op: context.op, + ...carried, + ...(context.noShare + ? { noShare: true as const } + : {}), + }, + }); + }; + if (!value) { + forward([...regions]); + return; + } + forward([...regions].filter((region) => !asking.has(region))); + forward( + [...regions].filter((region) => asking.has(region)), + value, + ); }); if (local.length === 0) return false; diff --git a/src/backend/services/events/forwarding.test.ts b/src/backend/services/events/forwarding.test.ts index dcb93d340..8382bf1d2 100644 --- a/src/backend/services/events/forwarding.test.ts +++ b/src/backend/services/events/forwarding.test.ts @@ -48,7 +48,11 @@ import type { FSEntry } from '../../stores/fs/FSEntry.js'; import type { IConfig } from '../../types.js'; import { EventForwardService } from './EventForwardService.js'; import { EventsService, type DeliveryEnvelope } from './EventsService.js'; -import type { ForwardBatch, ForwardDelivery } from './forwardQueue.js'; +import type { + ForwardBatch, + ForwardDelivery, + ForwardEvent, +} from './forwardQueue.js'; import { fsAnchorToken } from './subjects.js'; import type { WorkerInvocation, @@ -1922,6 +1926,126 @@ describe('forward-path metrics', () => { /** Session forwarding is on by default, so this is the ordinary config. */ const forwardCfg = { events: { enabled: true } } as Partial; +describe('a kv value crossing to another region', () => { + const APP = 'app-cart'; + + const subscribeKv = async ( + region: Region, + socketId: string, + includeValue?: true, + ) => + ( + await region.events.subscribe(actorFor(), socketId, { + subject: `kv:${APP}:cart`, + ...(includeValue ? { includeValue } : {}), + }) + ).sub.subId; + + const write = (region: Region) => + region.events.dispatchKv({ + userId, + namespace: `v1:user-${userId}:${APP}`, + keys: ['cart'], + op: 'set', + values: [{ items: 3 }], + }); + + /** The raw events a region forwarded, in order. */ + const eventsIn = (region: Region): ForwardEvent[] => + region.posts.flatMap((post) => + post.batch.items.filter( + (item): item is ForwardEvent => item.kind === 'event', + ), + ); + + /** Let `region` post past `before`, then let what it queued land. */ + const announced = async (region: Region, before = 0): Promise => { + await posted(region, before + 1); + await quiet(100); + }; + + it('goes nowhere a row did not ask for it', async () => { + const west = makeRegion('west', ['east'], forwardCfg); + const east = makeRegion('east', ['west'], forwardCfg); + await subscribeKv(east, 'socket-plain'); + await announced(east); + + await write(west); + await posted(west); + await arrived(east); + + expect(eventsIn(west)).toHaveLength(1); + expect(eventsIn(west)[0].kv).not.toHaveProperty('value'); + expect(east.sent[0].event).not.toHaveProperty('value'); + }); + + it('goes to a region holding a row that asked for it', async () => { + const west = makeRegion('west', ['east'], forwardCfg); + const east = makeRegion('east', ['west'], forwardCfg); + await subscribeKv(east, 'socket-valued', true); + await announced(east); + + await write(west); + await posted(west); + await arrived(east); + + expect(eventsIn(west)[0].kv).toMatchObject({ value: { items: 3 } }); + expect(east.sent[0].event).toMatchObject({ value: { items: 3 } }); + }); + + it('stops once the last row asking for it is unsubscribed', async () => { + const west = makeRegion('west', ['east'], forwardCfg); + const east = makeRegion('east', ['west'], forwardCfg); + const valued = await subscribeKv(east, 'socket-valued', true); + await subscribeKv(east, 'socket-plain'); + await announced(east); + const before = east.posts.length; + + await east.events.unsubscribe(actorFor(), 'socket-valued', { + subId: valued, + }); + await announced(east, before); + await write(west); + await posted(west); + + expect(eventsIn(west)).toHaveLength(1); + expect(eventsIn(west)[0].kv).not.toHaveProperty('value'); + }); + + it('stops once the connection holding it is reaped', async () => { + const west = makeRegion('west', ['east'], forwardCfg); + const east = makeRegion('east', ['west'], forwardCfg); + await subscribeKv(east, 'socket-valued', true); + await subscribeKv(east, 'socket-plain'); + await announced(east); + const before = east.posts.length; + + await east.events.reapSocket(userId, 'socket-valued'); + await announced(east, before); + await write(west); + await posted(west); + + expect(eventsIn(west)[0].kv).not.toHaveProperty('value'); + }); + + it('keeps going while another row there still asks for it', async () => { + const west = makeRegion('west', ['east'], forwardCfg); + const east = makeRegion('east', ['west'], forwardCfg); + const valued = await subscribeKv(east, 'socket-valued', true); + await subscribeKv(east, 'socket-also-valued', true); + await announced(east); + + await east.events.unsubscribe(actorFor(), 'socket-valued', { + subId: valued, + }); + await quiet(100); + await write(west); + await posted(west); + + expect(eventsIn(west)[0].kv).toMatchObject({ value: { items: 3 } }); + }); +}); + describe('a session subscription in another region', () => { it('delivers a write committed in the region that has no rows', async () => { const west = makeRegion('west', ['east'], forwardCfg); diff --git a/src/backend/stores/events/EventSubscriptionStore.test.ts b/src/backend/stores/events/EventSubscriptionStore.test.ts index e6c46b181..3cf9daafb 100644 --- a/src/backend/stores/events/EventSubscriptionStore.test.ts +++ b/src/backend/stores/events/EventSubscriptionStore.test.ts @@ -25,6 +25,7 @@ import type { IConfig } from '../../types.js'; import { EventSubscriptionStore, SESSION_SUBSCRIPTION_TTL_SECONDS, + valueWatchToken, type DurableSubscription, type SessionSubscription, } from './EventSubscriptionStore.js'; @@ -762,6 +763,109 @@ describe('the watched-set race', () => { }); }); +describe('rows asking for kv values', () => { + const VALUE = valueWatchToken('f#anchor'); + const asking = (over: Partial = {}) => + makeSub({ includeValue: true, ...over }); + + it('announces the value watch with the first row asking, and not again', async () => { + const first = await store.add(asking()); + const second = await store.add(asking({ socketId: 'socket-b' })); + + expect(first.announce).toEqual([ + { token: 'f#anchor', op: 'add' }, + { token: VALUE, op: 'add' }, + ]); + expect(second.announce).toBeUndefined(); + }); + + it('never announces one for a row that did not ask', async () => { + const bump = await store.add(makeSub()); + + expect(bump.announce).toEqual([{ token: 'f#anchor', op: 'add' }]); + }); + + it('withdraws it with the last row asking, and keeps the plain watch', async () => { + const valued = asking(); + await store.add(valued); + await store.add(makeSub({ socketId: 'socket-b' })); + + const bump = await store.remove(valued); + + expect(bump?.announce).toEqual([{ token: VALUE, op: 'drop' }]); + }); + + it('keeps it while another row still asks', async () => { + const valued = asking(); + await store.add(valued); + await store.add(asking({ socketId: 'socket-b' })); + + const bump = await store.remove(valued); + + expect(bump?.announce).toBeUndefined(); + }); + + it('withdraws it when the socket holding the row is reaped', async () => { + await store.add(asking()); + await store.add(makeSub({ socketId: 'socket-b' })); + + const [bump] = await store.reapSocket(USER, 'socket-a'); + + expect(bump.announce).toEqual([{ token: VALUE, op: 'drop' }]); + }); + + it('moves it with a row carried to another anchor', async () => { + const valued = asking(); + await store.add(valued); + + const { moved, bumps } = await store.reanchorSession(valued, { + ...valued, + token: 'f#parent', + anchorUid: 'parent', + anchorPath: '/testuser', + }); + + expect(moved).toBe(true); + expect(bumps[0].announce).toEqual( + expect.arrayContaining([ + { token: 'f#anchor', op: 'drop' }, + { token: VALUE, op: 'drop' }, + { token: 'f#parent', op: 'add' }, + { token: valueWatchToken('f#parent'), op: 'add' }, + ]), + ); + }); + + it('reasserts it on refresh only for a socket whose rows ask', async () => { + await store.add(asking()); + await store.add(makeSub({ socketId: 'socket-b' })); + + expect(await store.refresh(USER, 'socket-a')).toEqual([ + { ownerUserId: USER, token: 'f#anchor' }, + { ownerUserId: USER, token: VALUE }, + ]); + expect(await store.refresh(USER, 'socket-b')).toEqual([ + { ownerUserId: USER, token: 'f#anchor' }, + ]); + }); + + it('answers which peers asked for values alongside which watch', async () => { + await store.noteRemoteWatch(USER, 'f#anchor', 'east', 'add'); + await store.noteRemoteWatch(USER, 'f#anchor', 'south', 'add'); + await store.noteRemoteWatch(USER, VALUE, 'east', 'add'); + + const watched = await store.watchedFor(USER, ['f#anchor'], { + values: true, + }); + + expect(watched.remote.get('f#anchor')?.sort()).toEqual([ + 'east', + 'south', + ]); + expect(watched.remoteValues.get('f#anchor')).toEqual(['east']); + }); +}); + describe('reading rows for dispatch', () => { const durableRows = (count: number): DurableSubscription[] => Array.from( diff --git a/src/backend/stores/events/EventSubscriptionStore.ts b/src/backend/stores/events/EventSubscriptionStore.ts index b6fd88cfb..50c8dd8bf 100644 --- a/src/backend/stores/events/EventSubscriptionStore.ts +++ b/src/backend/stores/events/EventSubscriptionStore.ts @@ -82,6 +82,10 @@ import type { * into _its own_ `ev:rw`: which regions currently have a session watcher on one * of this owner's tokens. A write by this owner reads `ev:rw` alongside `ev:w` * to decide which peers, if any, get the raw event forwarded to them. + * + * A token that session rows here ask KV values on is counted and announced a + * second time, as its value variant, so a write sends the value only to the + * regions that asked for one. */ export type { @@ -105,6 +109,12 @@ const sessionCountKey = (userId: number | string): string => `ev:sc:{${userId}}`; const remoteWatchKey = (userId: number | string): string => `ev:rw:{${userId}}`; +/** + * The `ev:sc`/`ev:rw` field for session rows on `token` that ask for KV values. + * No anchor token starts with `v#`, so it can never name one. + */ +export const valueWatchToken = (token: string): string => `v#${token}`; + const safeParseRegions = (raw: string): Record | null => { try { return JSON.parse(raw) as Record; @@ -113,6 +123,30 @@ const safeParseRegions = (raw: string): Record | null => { } }; +/** Peers whose announcement of each token is still inside its TTL. */ +const liveRegionsByToken = ( + tokens: readonly string[], + raw: unknown, + cutoffMs: number, +): Map => { + const announced = (raw as Array | undefined) ?? []; + const live = new Map(); + tokens.forEach((token, i) => { + const entry = announced[i]; + if (!entry) return; + const regions = safeParseRegions(entry); + if (!regions) return; + const fresh = Object.entries(regions) + .filter( + ([, announcedAt]) => + typeof announcedAt === 'number' && announcedAt >= cutoffMs, + ) + .map(([region]) => region); + if (fresh.length > 0) live.set(token, fresh); + }); + return live; +}; + /** A row we cannot read is a row we cannot deliver against. */ const parseRow = (raw: string): DispatchSubscription | null => { try { @@ -303,23 +337,29 @@ export class EventSubscriptionStore extends PuterStore { throw counted?.[0] ?? subscriptionLimitReached(); } - const sessionCount = await this.#writeRow(sub); + const counts = await this.#writeRow(sub); await this.#keepDurableWindow(ownerUserId, [token]); + const announce: RemoteWatchAnnounce[] = []; + if (counts.rows === 1) announce.push({ token, op: 'add' }); + if (counts.values === 1) + announce.push({ token: valueWatchToken(token), op: 'add' }); return { userId: ownerUserId, generation: await this.bumpGeneration(ownerUserId), - announce: sessionCount === 1 ? [{ token, op: 'add' }] : undefined, + announce: announce.length > 0 ? announce : undefined, }; } /** - * Write one row's hash entry, watched-set membership and session count. - * Returns the resulting count for that token. Shared by `add` and - * `reanchorSession`, which both need the row in place before any ref names - * it. + * Write one row's hash entry, watched-set membership and session counts. + * Returns the resulting counts for that token — `values` only for a row + * asking for KV values. Shared by `add` and `reanchorSession`, which both + * need the row in place before any ref names it. */ - async #writeRow(sub: SessionSubscription): Promise { + async #writeRow( + sub: SessionSubscription, + ): Promise<{ rows: number; values: number | null }> { const { ownerUserId, token, subId } = sub; const rows = this.clients.redis.pipeline(); rows.hset(tokenKey(ownerUserId, token), subId, JSON.stringify(sub)); @@ -330,12 +370,21 @@ export class EventSubscriptionStore extends PuterStore { rows.sadd(watchedKey(ownerUserId), token); rows.expire(watchedKey(ownerUserId), SESSION_SUBSCRIPTION_TTL_SECONDS); rows.hincrby(sessionCountKey(ownerUserId), token, 1); + if (sub.includeValue) + rows.hincrby( + sessionCountKey(ownerUserId), + valueWatchToken(token), + 1, + ); rows.expire( sessionCountKey(ownerUserId), SESSION_SUBSCRIPTION_TTL_SECONDS, ); const results = (await rows.exec()) ?? []; - return Number(results[4]?.[1]); + return { + rows: Number(results[4]?.[1]), + values: sub.includeValue ? Number(results[5]?.[1]) : null, + }; } /** @@ -409,7 +458,7 @@ export class EventSubscriptionStore extends PuterStore { previous: SessionSubscription, next: SessionSubscription, ): Promise { - const nextCount = await this.#writeRow(next); + const nextCounts = await this.#writeRow(next); const swapped = await this.#scripts().eventsSwapSocketRef( socketKey(previous.holderUserId, previous.socketId), @@ -431,38 +480,50 @@ export class EventSubscriptionStore extends PuterStore { // first — undo the row this call wrote speculatively. 2: another // settle racing the same move already landed this exact ref; // its row is the one in place, so only this call's own count - // increment needs undoing. + // increments need undoing. if (swapped === 0) await this.#dropRows(next.ownerUserId, [ { token: next.token, subId: next.subId }, ]); - const undone = - (await this.#dropSessionCounts(next.ownerUserId, [next.token])) - .length > 0; - const announced = nextCount === 1; - if (announced === undone) return { moved: false, bumps: [] }; + const counted: Array<[string, number]> = [ + [next.token, nextCounts.rows], + ]; + if (nextCounts.values !== null) + counted.push([valueWatchToken(next.token), nextCounts.values]); + const undone = new Set( + ( + await this.#dropSessionCounts( + next.ownerUserId, + counted.map(([token]) => token), + ) + ).map((entry) => entry.token), + ); + const announce = counted.flatMap( + ([token, count]): RemoteWatchAnnounce[] => { + const announced = count === 1; + if (announced === undone.has(token)) return []; + return [{ token, op: announced ? 'add' : 'drop' }]; + }, + ); + if (announce.length === 0) return { moved: false, bumps: [] }; return { moved: false, bumps: [ { userId: next.ownerUserId, generation: await this.bumpGeneration(next.ownerUserId), - announce: [ - { - token: next.token, - op: announced ? 'add' : 'drop', - }, - ], + announce, }, ], }; } - await this.#dropRows(previous.ownerUserId, [ + const valued = await this.#dropRows(previous.ownerUserId, [ { token: previous.token, subId: previous.subId }, ]); const dropped = await this.#dropSessionCounts(previous.ownerUserId, [ previous.token, + ...valued, ]); await this.#keepDurableWindow(next.ownerUserId, [next.token]); @@ -472,10 +533,14 @@ export class EventSubscriptionStore extends PuterStore { ...(announceByOwner.get(ownerUserId) ?? []), { token, op: 'drop' }, ]); - if (nextCount === 1) + const added: RemoteWatchAnnounce[] = []; + if (nextCounts.rows === 1) added.push({ token: next.token, op: 'add' }); + if (nextCounts.values === 1) + added.push({ token: valueWatchToken(next.token), op: 'add' }); + if (added.length > 0) announceByOwner.set(next.ownerUserId, [ ...(announceByOwner.get(next.ownerUserId) ?? []), - { token: next.token, op: 'add' }, + ...added, ]); const owners = new Set([previous.ownerUserId, next.ownerUserId]); @@ -553,12 +618,12 @@ export class EventSubscriptionStore extends PuterStore { const dropped: DroppedSessionToken[] = []; for (const [ownerUserId, owned] of byOwner(removed)) { - await this.#dropRows(ownerUserId, owned); + const valued = await this.#dropRows(ownerUserId, owned); dropped.push( - ...(await this.#dropSessionCounts( - ownerUserId, - owned.map((ref) => ref.token), - )), + ...(await this.#dropSessionCounts(ownerUserId, [ + ...owned.map((ref) => ref.token), + ...valued, + ])), ); } return { dropped, removed }; @@ -605,17 +670,30 @@ export class EventSubscriptionStore extends PuterStore { * gone. Emptiness is what un-watches a token, which is what keeps one * socket's unsubscribe — or a durable row's removal — from silencing * another subscription on the same anchor. + * + * Returns the value variant of each token a dropped session row asked KV + * values on, read off the row as it goes. */ async #dropRows( ownerUserId: number, rows: ReadonlyArray<{ token: string; subId: string }>, - ): Promise { - if (rows.length === 0) return; + ): Promise { + if (rows.length === 0) return []; const drop = this.clients.redis.pipeline(); - for (const { token, subId } of rows) + for (const { token, subId } of rows) { + drop.hget(tokenKey(ownerUserId, token), subId); drop.hdel(tokenKey(ownerUserId, token), subId); - await drop.exec(); + } + const dropResults = (await drop.exec()) ?? []; + const valued = rows.flatMap(({ token }, i) => { + const raw = dropResults[2 * i]?.[1]; + if (Number(dropResults[2 * i + 1]?.[1]) !== 1) return []; + const row = typeof raw === 'string' ? parseRow(raw) : null; + return row?.socketId !== undefined && row.includeValue === true + ? [valueWatchToken(token)] + : []; + }); const tokens = [...new Set(rows.map((row) => row.token))]; const counts = this.clients.redis.pipeline(); @@ -625,7 +703,7 @@ export class EventSubscriptionStore extends PuterStore { const orphaned = tokens.filter( (_token, i) => Number(results[i]?.[1] ?? 0) === 0, ); - if (orphaned.length === 0) return; + if (orphaned.length === 0) return valued; await this.clients.redis.srem(watchedKey(ownerUserId), ...orphaned); @@ -642,6 +720,7 @@ export class EventSubscriptionStore extends PuterStore { ); if (revived.length > 0) await this.clients.redis.sadd(watchedKey(ownerUserId), ...revived); + return valued; } /** @@ -653,10 +732,10 @@ export class EventSubscriptionStore extends PuterStore { * dropped it (see `#dropRows`) heals itself on the next refresh even if * nothing catches it sooner. * - * Returns every (owner, token) pair the socket still holds, so the caller - * can re-announce them to peers — the whole of how a remote-watch - * announcement survives longer than one refresh window without a second - * timer. + * Returns every (owner, token) pair the socket still holds, and the value + * variant of each one a row asks KV values on, so the caller can + * re-announce them to peers — the whole of how a remote-watch announcement + * survives longer than one refresh window without a second timer. */ async refresh( holderUserId: number, @@ -689,10 +768,23 @@ export class EventSubscriptionStore extends PuterStore { tokenKey(ownerUserId, token), SESSION_SUBSCRIPTION_TTL_SECONDS, ); - await pipeline.exec(); + for (const ref of owned) + pipeline.hget(tokenKey(ownerUserId, ref.token), ref.subId); + const results = (await pipeline.exec()) ?? []; + + const rowsAt = 3 + tokens.length; + const valued = new Set( + owned.flatMap((ref, i) => { + const raw = results[rowsAt + i]?.[1]; + const row = typeof raw === 'string' ? parseRow(raw) : null; + return row?.includeValue === true ? [ref.token] : []; + }), + ); await this.#keepDurableWindow(ownerUserId, tokens); for (const token of tokens) reasserted.push({ ownerUserId, token }); + for (const token of valued) + reasserted.push({ ownerUserId, token: valueWatchToken(token) }); } return reasserted; } @@ -827,37 +919,40 @@ export class EventSubscriptionStore extends PuterStore { async watchedFor( ownerUserId: number, tokens: readonly string[], - ): Promise<{ local: string[]; remote: Map }> { - if (tokens.length === 0) return { local: [], remote: new Map() }; + options: { values?: boolean } = {}, + ): Promise<{ + local: string[]; + remote: Map; + /** + * With `values`: the regions whose rows on each token ask for KV + * values. + */ + remoteValues: Map; + }> { + if (tokens.length === 0) + return { local: [], remote: new Map(), remoteValues: new Map() }; const pipeline = this.clients.redis.pipeline(); pipeline.smismember(watchedKey(ownerUserId), ...tokens); pipeline.hmget(remoteWatchKey(ownerUserId), ...tokens); + if (options.values) + pipeline.hmget( + remoteWatchKey(ownerUserId), + ...tokens.map(valueWatchToken), + ); const results = (await pipeline.exec()) ?? []; const flags = (results[0]?.[1] as number[] | undefined) ?? []; const local = tokens.filter((_token, i) => Number(flags[i]) === 1); - const rawRemote = - (results[1]?.[1] as Array | undefined) ?? []; const cutoffMs = Date.now() - REMOTE_WATCH_TTL_SECONDS * 1000; - const remote = new Map(); - tokens.forEach((token, i) => { - const raw = rawRemote[i]; - if (!raw) return; - const regions = safeParseRegions(raw); - if (!regions) return; - const live = Object.entries(regions) - .filter( - ([, announcedAt]) => - typeof announcedAt === 'number' && - announcedAt >= cutoffMs, - ) - .map(([region]) => region); - if (live.length > 0) remote.set(token, live); - }); - - return { local, remote }; + return { + local, + remote: liveRegionsByToken(tokens, results[1]?.[1], cutoffMs), + remoteValues: options.values + ? liveRegionsByToken(tokens, results[2]?.[1], cutoffMs) + : new Map(), + }; } /**