openvibers/openvibe.report Commits
Report's Live reads: serve an expired answer at once and refresh it behind the page (#72)
Changes
@@@ -7,7 +7,9 @@** const cache = createCache({ max: 1000, now: () => Date.now() });* cache.get(key) undefined when absent or expired++ * cache.peek(key, staleMs) the value even past its expiry, while it expired under staleMs ago (else undefined)* cache.set(key, value, ttlMs)++ * An expired entry stays until it is replaced or pushed out, so a client can serve it while it refreshes.*/function createCache({ max = 1000, now = () => Date.now() } = {}) {const entries = new Map();@@@ -16,10 +18,7 @@ function createCache({ max = 1000, now = () => Date.now() } = {}) {function get(key) {const hit = entries.get(key);if (!hit) return undefined;-- if (hit.expiresAt <= clock()) {-- entries.delete(key);-- return undefined;-- }++ if (hit.expiresAt <= clock()) return undefined;// Do not move it to the end: "oldest out" means insertion order, not least-recently-used.return hit.value;}@@@ -34,8 +33,15 @@ function createCache({ max = 1000, now = () => Date.now() } = {}) {return value;}++ function peek(key, staleMs = 0) {++ const hit = entries.get(key);++ if (!hit || clock() - hit.expiresAt >= Math.max(0, Number(staleMs) || 0)) return undefined;++ return hit.value;++ }++return {get,++ peek,set,delete: (key) => entries.delete(key),get size() { return entries.size; },@@@ -17,6 +17,8 @@const { createCache } = require('./cache');const { trim, requestJson, isSlug, streamIdOf, isUsername, TIMEOUT_MS } = require('./source');++const SERVE_STALE_MS = 30 * 60_000;++const TTL = {rundown: 2 * 60_000,topics: 5 * 60_000,@@@ -34,13 +36,25 @@ function createLiveClient({ config = {}, fetchImpl = globalThis.fetch, log = conconst baseUrl = trim(config.liveUrl || config.liveInternalUrl || 'http://127.0.0.1:3000');const cache = createCache({ max: cacheMax, now });-- async function cached(key, path, ttl, label) {-- const hit = cache.get(key);-- if (hit !== undefined) return hit;++ // An answer past its time but expired under SERVE_STALE_MS ago is served at once while one refresh per key runs++ // behind the page: the visitor after an expiry waited for Live (/rundown at up to 1.9 s at the origin, 11-02). A++ // failed refresh leaves the stale answer to be served (and refreshed again) until it is too old.++ const refreshing = new Map();++ async function load(key, path, ttl, label) {const data = await requestJson(`${baseUrl}${path}`, { fetchImpl, log, timeoutMs, label });if (data === null) return null;return cache.set(key, data, ttl);}++ async function cached(key, path, ttl, label) {++ const hit = cache.get(key);++ if (hit !== undefined) return hit;++ const stale = cache.peek(key, SERVE_STALE_MS);++ if (stale !== undefined) {++ if (!refreshing.has(key)) refreshing.set(key, load(key, path, ttl, label).catch(() => null).finally(() => refreshing.delete(key)));++ return stale;++ }++ return load(key, path, ttl, label);++ }function rundown() {return cached('rundown:home', '/api/rundown/', TTL.rundown, 'live rundown');@@@ -80,7 +94,7 @@ function createLiveClient({ config = {}, fetchImpl = globalThis.fetch, log = conreturn cached(`recaps:channel:${key}`, `/api/recap/channel/${encodeURIComponent(username)}`, TTL.channelRecaps, 'live channel recaps');}-- return { rundown, topics, topic, episode, speaker, recap, channelRecaps, cache, baseUrl, ttlMs: TTL };++ return { rundown, topics, topic, episode, speaker, recap, channelRecaps, cache, baseUrl, ttlMs: TTL, settled: () => Promise.all([...refreshing.values()]) };}module.exports = { createLiveClient, TTL };@@@ -136,6 +136,21 @@ const quiet = () => ({ warn() {}, error() {} });assert.strictEqual(urls.filter((u) => u.includes('/api/rundown/s/')).length, before);});++ await check('live: an expired answer is served at once and refreshed behind the page, one refresh for three readers', async () => {++ let now = 0;++ let n = 0;++ const client = createLiveClient({ config: { liveUrl: 'http://live' }, fetchImpl: async () => jsonResponse({ n: ++n }), now: () => now, log: quiet() });++ assert.strictEqual((await client.rundown()).n, 1);++ now += 3 * 60_000;++ const three = await Promise.all([client.rundown(), client.rundown(), client.rundown()]);++ assert.deepStrictEqual(three.map((x) => x.n), [1, 1, 1], 'the last answer at once');++ await client.settled();++ assert.strictEqual(n, 2, 'one refresh behind the page');++ assert.strictEqual((await client.rundown()).n, 2, 'the refreshed answer on the next read');++ now += 60 * 60_000;++ assert.strictEqual((await client.rundown()).n, 3, 'too old to serve: fetched while the visitor waits');++ });++await check('the cache keeps 1000 entries and drops the oldest first', async () => {let calls = 0;const client = createLiveClient({
@@ -7,7 +7,9 @@
@@ -7,7 +7,9 @@
*
*
* const cache = createCache({ max: 1000, now: () => Date.now() });
* const cache = createCache({ max: 1000, now: () => Date.now() });
* cache.get(key) undefined when absent or expired
* cache.get(key) undefined when absent or expired
+ * cache.peek(key, staleMs) the value even past its expiry, while it expired under staleMs ago (else undefined)
* cache.set(key, value, ttlMs)
* cache.set(key, value, ttlMs)
+ * An expired entry stays until it is replaced or pushed out, so a client can serve it while it refreshes.
*/
*/
function createCache({ max = 1000, now = () => Date.now() } = {}) {
function createCache({ max = 1000, now = () => Date.now() } = {}) {
const entries = new Map();
const entries = new Map();
@@ -16,10 +18,7 @@ function createCache({ max = 1000, now = () => Date.now() } = {}) {
@@ -16,10 +18,7 @@ function createCache({ max = 1000, now = () => Date.now() } = {}) {
function get(key) {
function get(key) {
const hit = entries.get(key);
const hit = entries.get(key);
if (!hit) return undefined;
if (!hit) return undefined;
- if (hit.expiresAt <= clock()) {
- entries.delete(key);
- return undefined;
- }
+ if (hit.expiresAt <= clock()) return undefined;
// Do not move it to the end: "oldest out" means insertion order, not least-recently-used.
// Do not move it to the end: "oldest out" means insertion order, not least-recently-used.
return hit.value;
return hit.value;
}
}
@@ -34,8 +33,15 @@ function createCache({ max = 1000, now = () => Date.now() } = {}) {
@@ -34,8 +33,15 @@ function createCache({ max = 1000, now = () => Date.now() } = {}) {
return value;
return value;
}
}
+ function peek(key, staleMs = 0) {
+ const hit = entries.get(key);
+ if (!hit || clock() - hit.expiresAt >= Math.max(0, Number(staleMs) || 0)) return undefined;
+ return hit.value;
+ }
+
return {
return {
get,
get,
+ peek,
set,
set,
delete: (key) => entries.delete(key),
delete: (key) => entries.delete(key),
get size() { return entries.size; },
get size() { return entries.size; },
@@ -17,6 +17,8 @@
@@ -17,6 +17,8 @@
const { createCache } = require('./cache');
const { createCache } = require('./cache');
const { trim, requestJson, isSlug, streamIdOf, isUsername, TIMEOUT_MS } = require('./source');
const { trim, requestJson, isSlug, streamIdOf, isUsername, TIMEOUT_MS } = require('./source');
+const SERVE_STALE_MS = 30 * 60_000;
+
const TTL = {
const TTL = {
rundown: 2 * 60_000,
rundown: 2 * 60_000,
topics: 5 * 60_000,
topics: 5 * 60_000,
@@ -34,13 +36,25 @@ function createLiveClient({ config = {}, fetchImpl = globalThis.fetch, log = con
@@ -34,13 +36,25 @@ function createLiveClient({ config = {}, fetchImpl = globalThis.fetch, log = con
const baseUrl = trim(config.liveUrl || config.liveInternalUrl || 'http://127.0.0.1:3000');
const baseUrl = trim(config.liveUrl || config.liveInternalUrl || 'http://127.0.0.1:3000');
const cache = createCache({ max: cacheMax, now });
const cache = createCache({ max: cacheMax, now });
- async function cached(key, path, ttl, label) {
- const hit = cache.get(key);
- if (hit !== undefined) return hit;
+ // An answer past its time but expired under SERVE_STALE_MS ago is served at once while one refresh per key runs
+ // behind the page: the visitor after an expiry waited for Live (/rundown at up to 1.9 s at the origin, 11-02). A
+ // failed refresh leaves the stale answer to be served (and refreshed again) until it is too old.
+ const refreshing = new Map();
+ async function load(key, path, ttl, label) {
const data = await requestJson(`${baseUrl}${path}`, { fetchImpl, log, timeoutMs, label });
const data = await requestJson(`${baseUrl}${path}`, { fetchImpl, log, timeoutMs, label });
if (data === null) return null;
if (data === null) return null;
return cache.set(key, data, ttl);
return cache.set(key, data, ttl);
}
}
+ async function cached(key, path, ttl, label) {
+ const hit = cache.get(key);
+ if (hit !== undefined) return hit;
+ const stale = cache.peek(key, SERVE_STALE_MS);
+ if (stale !== undefined) {
+ if (!refreshing.has(key)) refreshing.set(key, load(key, path, ttl, label).catch(() => null).finally(() => refreshing.delete(key)));
+ return stale;
+ }
+ return load(key, path, ttl, label);
+ }
function rundown() {
function rundown() {
return cached('rundown:home', '/api/rundown/', TTL.rundown, 'live rundown');
return cached('rundown:home', '/api/rundown/', TTL.rundown, 'live rundown');
@@ -80,7 +94,7 @@ function createLiveClient({ config = {}, fetchImpl = globalThis.fetch, log = con
@@ -80,7 +94,7 @@ function createLiveClient({ config = {}, fetchImpl = globalThis.fetch, log = con
return cached(`recaps:channel:${key}`, `/api/recap/channel/${encodeURIComponent(username)}`, TTL.channelRecaps, 'live channel recaps');
return cached(`recaps:channel:${key}`, `/api/recap/channel/${encodeURIComponent(username)}`, TTL.channelRecaps, 'live channel recaps');
}
}
- return { rundown, topics, topic, episode, speaker, recap, channelRecaps, cache, baseUrl, ttlMs: TTL };
+ return { rundown, topics, topic, episode, speaker, recap, channelRecaps, cache, baseUrl, ttlMs: TTL, settled: () => Promise.all([...refreshing.values()]) };
}
}
module.exports = { createLiveClient, TTL };
module.exports = { createLiveClient, TTL };
@@ -136,6 +136,21 @@ const quiet = () => ({ warn() {}, error() {} });
@@ -136,6 +136,21 @@ const quiet = () => ({ warn() {}, error() {} });
assert.strictEqual(urls.filter((u) => u.includes('/api/rundown/s/')).length, before);
assert.strictEqual(urls.filter((u) => u.includes('/api/rundown/s/')).length, before);
});
});
+ await check('live: an expired answer is served at once and refreshed behind the page, one refresh for three readers', async () => {
+ let now = 0;
+ let n = 0;
+ const client = createLiveClient({ config: { liveUrl: 'http://live' }, fetchImpl: async () => jsonResponse({ n: ++n }), now: () => now, log: quiet() });
+ assert.strictEqual((await client.rundown()).n, 1);
+ now += 3 * 60_000;
+ const three = await Promise.all([client.rundown(), client.rundown(), client.rundown()]);
+ assert.deepStrictEqual(three.map((x) => x.n), [1, 1, 1], 'the last answer at once');
+ await client.settled();
+ assert.strictEqual(n, 2, 'one refresh behind the page');
+ assert.strictEqual((await client.rundown()).n, 2, 'the refreshed answer on the next read');
+ now += 60 * 60_000;
+ assert.strictEqual((await client.rundown()).n, 3, 'too old to serve: fetched while the visitor waits');
+ });
+
await check('the cache keeps 1000 entries and drops the oldest first', async () => {
await check('the cache keeps 1000 entries and drops the oldest first', async () => {
let calls = 0;
let calls = 0;
const client = createLiveClient({
const client = createLiveClient({