diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md index 710b92f..4cc36ab 100644 --- a/ACCEPTANCE.md +++ b/ACCEPTANCE.md @@ -2189,6 +2189,63 @@ green — the fallback layer is retained, now reached only per DM-A3. --- +## Bugfix: Progressive backfill for every bulk collection (issue #832) + +The server pages **every** bulk collection backward via `?before=` (SPEC +BP1-BP8), but only the message feed wired it on the client (the deferred +follow-up **BP9a**). So the node table — and positions, telemetry, neighbors, +traces — stalled at the newest `MAX_QUERY_LIMIT` (1000) rows the server returns +in one page (the reported symptom: "the node table only lists 1000 items"). +The fix mirrors the proven chat backfill (issue #802) across all five +collections: the newest page paints first, then a one-shot background pager +walks each collection's inclusive `before` cursor newest → oldest, de-duplicating +by id and committing+rendering each page, until the visibility window is +exhausted. The client row-caps on positions/telemetry/traces are lifted from a +fixed count to the server's own window bound (so a backfilled page is not trimmed +straight back out on the next refresh). Frontend-only: no API/DB/ingestor change, +so the C4/C7 window floors, `MAX_QUERY_LIMIT`, and privacy are untouched. + +### CB-A1 — Every bulk collection pages backward past the first 1000-row page +```bash +( cd web && node --test public/assets/js/app/__tests__/main-collection-backfill.test.js ) +``` +**Expected:** pass. On a cold load whose newest page is **full** (=== the +per-collection cap), each of `nodes`, `positions`, `telemetry`, `neighbors`, and +`traces` issues at least one `GET /api/?…&before=` request and +merges the older rows in — so the loaded node set grows **past** `NODE_LIMIT` +(1000) instead of stalling at it. The newest page is rendered **before** any +backward paging starts (the page is never blank/blocking), matching the #802 +progressive-load contract. A short newest page (window already exhausted) records +no frontier and fires **no** backward request. + +### CB-A2 — Generic backward pager + `before` cursor on every fetcher +```bash +( cd web && node --test public/assets/js/app/main/__tests__/data-fetchers.test.js ) +``` +**Expected:** pass. `paginateCollection(fetchPage, {limit, before, idOf, cursorOf})` +generalises the message walk (`paginateMessages` now delegates to it): it pages +newest → oldest, de-duplicates by `idOf`, advances an inclusive `before` cursor to +the oldest `cursorOf` value of each page, and stops on a short page / no-progress / +missing cursor / `maxPages`. `fetchNodes`/`fetchPositions`/`fetchTelemetry`/ +`fetchNeighbors`/`fetchTraces` each forward a positive `before` and omit a +non-positive one (mirroring the existing `fetchMessages` `before` contract, C7); +`fetchTraces` accepts `applyAgeFilter:false` so the pager sees the server's raw +page length and terminates correctly. + +### CB-R1 — Regression: prior acceptance still holds +```bash +( cd web && npm test ) && ( cd web && bundle exec rspec ) +``` +**Expected:** every prior check still passes. At risk and explicitly required to +stay green: **C7 / PL-A1 / PL-A2** (the message pager is unchanged — `paginateMessages` +delegates to the new generic pager with identical observable behavior), **B1** +(all suites), and **B4** (the exact Apache header on the new test). The cursor +columns match the server's `ORDER BY` per collection (`last_heard` for nodes, +`rx_time` for the rest), so no widening of the C4 window floor is possible; the +backfill only ever *narrows* (BP2). + +--- + ## Bugfix: MeshCore dedup window vs inter-ingestor clock skew; warm-cache chat gap Two chat defects found on production `potatomesh.net` (v0.7.1-rc0) with two diff --git a/web/public/assets/js/app/__tests__/main-collection-backfill.test.js b/web/public/assets/js/app/__tests__/main-collection-backfill.test.js new file mode 100644 index 0000000..be95bd8 --- /dev/null +++ b/web/public/assets/js/app/__tests__/main-collection-backfill.test.js @@ -0,0 +1,284 @@ +/* + * Copyright © 2025-26 l5yth & contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Regression guard for issue #832 (frontend data population). + * + * The server supports backward `?before=` pagination on every bulk collection + * (SPEC BP1-BP8), but only the message feed wired it on the client — so the + * node table (and positions / telemetry / neighbors / traces) stalled at the + * newest `MAX_QUERY_LIMIT` (1000) rows the server returns in one page. This + * test pins the fix (SPEC BP9a follow-up): after the newest page paints, every + * bulk collection pages backward through its visibility window in the + * background, so the table fills past the first 1000 rows. + * + * @module app/__tests__/main-collection-backfill + */ + +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { createDomEnvironment } from './dom-environment.js'; +import { initializeApp } from '../main.js'; +import { NODE_LIMIT, TRACE_LIMIT } from '../main/constants.js'; + +/** Minimal config that disables the auto-refresh timer so timing is ours. */ +const BASE_CONFIG = Object.freeze({ + channel: 'Primary', + frequency: '915MHz', + refreshMs: 0, + refreshIntervalSeconds: 0, + chatEnabled: true, + mapCenter: { lat: 0, lon: 0 }, + mapZoom: null, + maxDistanceKm: 0, + instancesFeatureEnabled: false, + instanceDomain: null, + snapshotWindowSeconds: 3600, +}); + +/** Build a resolved fetch-style response around a JSON body. */ +function jsonResponse(body) { + return Promise.resolve({ ok: true, status: 200, json: () => Promise.resolve(body) }); +} + +/** Yield to pending microtasks/timers so the background backfill settles. */ +function settle(ms = 120) { + return new Promise(r => setTimeout(r, ms)); +} + +/** Format an integer as a canonical `!%08x` node id. */ +const nid = n => `!${n.toString(16).padStart(8, '0')}`; + +test('every bulk collection pages backward past the first 1000-row page (#832)', async () => { + const now = Math.floor(Date.now() / 1000); + + // Full newest pages (=== the per-collection cap) so each backfill continues to + // an older page; a short page would (correctly) mean the window is exhausted. + const newestNodes = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + node_id: nid(0x10000 + i), last_heard: now - i, short_name: `N${i}`, role: 'CLIENT', + })); + const newestPositions = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + id: 1_000_000 + i, node_id: nid(0x10000 + i), rx_time: now - i, latitude: 52, longitude: 13, + })); + const newestTelemetry = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + id: 2_000_000 + i, node_id: nid(0x10000 + i), rx_time: now - i, battery_level: 50, + })); + const newestNeighbors = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + node_id: nid(0x10000 + i), neighbor_id: nid(0x10001 + i), rx_time: now - i, snr: 5, + })); + const newestTraces = Array.from({ length: TRACE_LIMIT }, (_, i) => ({ + id: 3_000_000 + i, rx_time: now - i, from_id: nid(0x10000 + i), to_id: nid(0x20000 + i), + })); + + // One older row per collection (a short page → the walk stops after one step). + const older = now - NODE_LIMIT - 50; + const olderNodes = [{ node_id: '!aaaa0001', last_heard: older, short_name: 'OLD', role: 'CLIENT' }]; + const olderPositions = [{ id: 1_900_000, node_id: '!aaaa0001', rx_time: older, latitude: 52, longitude: 13 }]; + const olderTelemetry = [{ id: 2_900_000, node_id: '!aaaa0001', rx_time: older, battery_level: 10 }]; + const olderNeighbors = [{ node_id: '!aaaa0001', neighbor_id: '!aaaa0002', rx_time: older, snr: 1 }]; + const olderTraces = [{ id: 3_900_000, rx_time: older, from_id: '!aaaa0001', to_id: '!aaaa0002' }]; + + const beforeRequested = { + nodes: false, positions: false, telemetry: false, neighbors: false, traces: false, + }; + + function stubFetch(url) { + // The hydrator must not fall back to per-node lookups (it resolves from the + // bulk map); answer 404-equivalent just in case. + if (url.startsWith('/api/nodes/')) return jsonResponse(null); + const isBefore = url.includes('before='); + if (url.startsWith('/api/nodes')) { + if (isBefore) { beforeRequested.nodes = true; return jsonResponse(olderNodes); } + return jsonResponse(newestNodes); + } + if (url.startsWith('/api/positions')) { + if (isBefore) { beforeRequested.positions = true; return jsonResponse(olderPositions); } + return jsonResponse(newestPositions); + } + if (url.startsWith('/api/telemetry')) { + if (isBefore) { beforeRequested.telemetry = true; return jsonResponse(olderTelemetry); } + return jsonResponse(newestTelemetry); + } + if (url.startsWith('/api/neighbors')) { + if (isBefore) { beforeRequested.neighbors = true; return jsonResponse(olderNeighbors); } + return jsonResponse(newestNeighbors); + } + if (url.startsWith('/api/traces')) { + if (isBefore) { beforeRequested.traces = true; return jsonResponse(olderTraces); } + return jsonResponse(newestTraces); + } + if (url.startsWith('/api/messages')) return jsonResponse([]); // chat empty → no interference + return jsonResponse([]); // /api/stats and anything else + } + + const env = createDomEnvironment({ includeBody: true }); + const originalFetch = globalThis.fetch; + globalThis.fetch = url => stubFetch(url); + try { + const { _testUtils } = initializeApp(BASE_CONFIG); + await _testUtils.initialLoad; + + // The newest page paints first — the table is filled to the server's cap + // before any backward paging happens (so the page is never blank/blocking). + assert.equal( + _testUtils.getLoadedNodeCount(), NODE_LIMIT, + 'the newest page must render before backward paging starts', + ); + + // Let the one-shot background backfill page each collection backward. + await _testUtils.flushCollectionBackfills(); + await settle(); + + // Every bulk collection paged backward via ?before= — the core of the fix. + assert.deepEqual( + beforeRequested, + { nodes: true, positions: true, telemetry: true, neighbors: true, traces: true }, + 'each bulk collection must page backward through its window, not stall at the newest 1000 rows', + ); + + // The node table grew past the single 1000-row page (the user-visible bug: + // "the node table only lists 1000 items"). + assert.equal( + _testUtils.getLoadedNodeCount(), NODE_LIMIT + 1, + `older nodes beyond the first page must be backfilled in (loaded=${_testUtils.getLoadedNodeCount()})`, + ); + // The other four collections also grew by their one older page. + assert.equal(_testUtils.getLoadedPositionCount(), NODE_LIMIT + 1, 'positions backfilled'); + assert.equal(_testUtils.getLoadedTelemetryCount(), NODE_LIMIT + 1, 'telemetry backfilled'); + assert.equal(_testUtils.getLoadedNeighborCount(), NODE_LIMIT + 1, 'neighbors backfilled'); + assert.equal(_testUtils.getLoadedTraceCount(), TRACE_LIMIT + 1, 'traces backfilled'); + } finally { + globalThis.fetch = originalFetch; + env.cleanup(); + } +}); + +test('a short newest page records no frontier and fires no backward request (#832)', async () => { + const now = Math.floor(Date.now() / 1000); + // Every newest page is *short* (< the per-collection cap) ⇒ the window is + // already exhausted, so the backfill must record no frontier and issue no + // ?before= request (no empty, long-loading page — the perf requirement). + const nodes = [{ node_id: '!aaaa0001', last_heard: now, short_name: 'A', role: 'CLIENT' }]; + const beforeSeen = []; + function stubFetch(url) { + if (url.startsWith('/api/nodes/')) return jsonResponse(null); + if (url.includes('before=')) beforeSeen.push(url); + if (url.startsWith('/api/nodes')) return jsonResponse(nodes); + if (url.startsWith('/api/messages')) return jsonResponse([]); + return jsonResponse([]); // positions / telemetry / neighbors / traces / stats short + } + + const env = createDomEnvironment({ includeBody: true }); + const originalFetch = globalThis.fetch; + globalThis.fetch = url => stubFetch(url); + try { + const { _testUtils } = initializeApp(BASE_CONFIG); + await _testUtils.initialLoad; + await _testUtils.flushCollectionBackfills(); + await settle(); + assert.deepEqual(beforeSeen, [], 'no collection should page backward when its window is not full'); + assert.equal(_testUtils.getLoadedNodeCount(), 1, 'the single short page is shown as-is'); + } finally { + globalThis.fetch = originalFetch; + env.cleanup(); + } +}); + +test('a failed backfill page is swallowed and leaves the newest page intact (#832)', async () => { + const now = Math.floor(Date.now() / 1000); + const newestNodes = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + node_id: nid(0x10000 + i), last_heard: now - i, short_name: `N${i}`, role: 'CLIENT', + })); + function stubFetch(url) { + if (url.startsWith('/api/nodes/')) return jsonResponse(null); + if (url.startsWith('/api/nodes')) { + if (url.includes('before=')) return Promise.reject(new Error('nodes backfill boom')); + return jsonResponse(newestNodes); + } + if (url.startsWith('/api/messages')) return jsonResponse([]); + return jsonResponse([]); + } + + const env = createDomEnvironment({ includeBody: true }); + const originalFetch = globalThis.fetch; + const originalWarn = console.warn; + const warnings = []; + console.warn = (...args) => warnings.push(args); + globalThis.fetch = url => stubFetch(url); + try { + const { _testUtils } = initializeApp(BASE_CONFIG); + await _testUtils.initialLoad; + await _testUtils.flushCollectionBackfills(); + await settle(); + // The rejected backward page is caught (not rethrown) and the newest page + // stays rendered. + assert.equal(_testUtils.getLoadedNodeCount(), NODE_LIMIT); + assert.ok( + warnings.some(args => String(args[0]).includes('nodes backfill failed')), + 'the backfill failure should be logged, not rethrown', + ); + } finally { + console.warn = originalWarn; + globalThis.fetch = originalFetch; + env.cleanup(); + } +}); + +test('the node backfill walks multiple older pages until the window is exhausted (#832)', async () => { + const now = Math.floor(Date.now() / 1000); + // Newest page (full) → first older page (full, strictly older) → short page. + const newestNodes = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + node_id: nid(0x10000 + i), last_heard: now - i, short_name: `N${i}`, role: 'CLIENT', + })); + const olderPage1 = Array.from({ length: NODE_LIMIT }, (_, i) => ({ + node_id: nid(0x80000 + i), last_heard: now - NODE_LIMIT - i, short_name: `O${i}`, role: 'CLIENT', + })); + const olderPage2 = [{ node_id: '!ffff0001', last_heard: now - 2 * NODE_LIMIT - 5, short_name: 'LAST', role: 'CLIENT' }]; + + let nodeBeforeCount = 0; + function stubFetch(url) { + if (url.startsWith('/api/nodes/')) return jsonResponse(null); + if (url.startsWith('/api/nodes')) { + if (url.includes('before=')) { + nodeBeforeCount += 1; + if (nodeBeforeCount === 1) return jsonResponse(olderPage1); // full → keep paging + if (nodeBeforeCount === 2) return jsonResponse(olderPage2); // short → stop + return jsonResponse([]); + } + return jsonResponse(newestNodes); + } + if (url.startsWith('/api/messages')) return jsonResponse([]); + return jsonResponse([]); + } + + const env = createDomEnvironment({ includeBody: true }); + const originalFetch = globalThis.fetch; + globalThis.fetch = url => stubFetch(url); + try { + const { _testUtils } = initializeApp(BASE_CONFIG); + await _testUtils.initialLoad; + await _testUtils.flushCollectionBackfills(); + await settle(); + assert.equal(nodeBeforeCount, 2, 'paged backward twice, then stopped on the short page'); + assert.equal( + _testUtils.getLoadedNodeCount(), NODE_LIMIT * 2 + 1, + `the whole window must be paged in (loaded=${_testUtils.getLoadedNodeCount()})`, + ); + } finally { + globalThis.fetch = originalFetch; + env.cleanup(); + } +}); diff --git a/web/public/assets/js/app/main.js b/web/public/assets/js/app/main.js index 4f24c96..7e355c4 100644 --- a/web/public/assets/js/app/main.js +++ b/web/public/assets/js/app/main.js @@ -177,6 +177,7 @@ import { resolveSnapshotLimit, fetchMessages as fetchMessagesImpl, paginateMessages as paginateMessagesImpl, + paginateCollection, } from './main/data-fetchers.js'; import { compareNumber, @@ -347,6 +348,18 @@ export function initializeApp(config) { let chatLiveFrontier = 0; /** Settles when the one-shot chat-history backfill finishes (test hook). */ let backfillPromise = Promise.resolve(); + /** + * Live frontiers for the bulk collections (oldest cursor value of the newest + * page) — the one-shot background backfill pages backward from here, exactly + * like {@link chatLiveFrontier}. 0 means "no backfill" (a short newest page, + * or a warm-cache load whose data is already seeded). Issue #832 / SPEC BP9a. + * @type {{ nodes: number, positions: number, telemetry: number, neighbors: number, traces: number }} + */ + let collectionLiveFrontiers = { nodes: 0, positions: 0, telemetry: 0, neighbors: 0, traces: 0 }; + /** One-shot guard: the bulk-collection backfill runs once after the first load. */ + let collectionsBackfilled = false; + /** Settles when the one-shot bulk-collection backfill finishes (test hook). */ + let collectionBackfillPromise = Promise.resolve(); // Persistent read-side cache (SPEC FC1–FC7). The IndexedDB backend is null // when storage is unavailable, and PRIVATE mode disables + wipes the cache — @@ -3488,6 +3501,178 @@ export function initializeApp(config) { } } + /** + * Re-aggregate the per-source snapshot arrays and re-enrich the node + * collection (display name, position, distance, telemetry) from the current + * module-level ``all*`` sources, then rebuild the node lookup index. Shared by + * {@link refresh} and the background collection backfill (issue #832) so a + * streamed history page derives the rendered node state identically to a full + * refresh. Does not touch ``allMessages`` / ``allEncryptedMessages`` (hydrated + * separately) or ``allTraces`` (not node-derived). + * + * @returns {void} + */ + function rebuildNodeDerivedState() { + const aggregatedNodes = aggregateNodeSnapshots(allNodes); + const aggregatedPositions = aggregatePositionSnapshots(allPositionEntries); + const aggregatedNeighbors = aggregateNeighborSnapshots(allNeighbors); + const aggregatedTelemetry = aggregateTelemetrySnapshots(allTelemetryEntries); + // Enrich merged node records with display name, position, distance, and + // telemetry before any rendering or filtering takes place. + aggregatedNodes.forEach(applyNodeNameFallback); + mergePositionsIntoNodes(aggregatedNodes, aggregatedPositions); + computeDistances(aggregatedNodes); + mergeTelemetryIntoNodes(aggregatedNodes, aggregatedTelemetry); + normalizeNodeCollection(aggregatedNodes); + allNodes = aggregatedNodes; + // Rebuild lookup maps so marker updates and message hydration always resolve + // to the latest node objects. + rebuildNodeIndex(allNodes); + allTelemetryEntries = aggregatedTelemetry; + allPositionEntries = aggregatedPositions; + allNeighbors = aggregatedNeighbors; + } + + /** Floor (unix s) below which backfilled positions/telemetry are dropped (FC3: 7 d). */ + const recentBackfillFloor = () => Math.floor(Date.now() / 1000) - CHAT_RECENT_WINDOW_SECONDS; + /** Floor (unix s) below which backfilled neighbors/traces are dropped (FC3: 28 d). */ + const longBackfillFloor = () => Math.floor(Date.now() / 1000) - TRACE_MAX_AGE_SECONDS; + + /** + * Per-collection wiring for the background backfill (issue #832). Each entry + * knows how to fetch a backward page (inclusive ``before`` cursor on the + * route's primary sort column), how to identify a row for cross-page + * de-duplication, which cursor value advances the walk, how to merge a page + * into the module state (bounded by the same window the refresh tick uses), and + * how to re-derive the rendered state the merged collection feeds. + * + * Cursor columns match each route's server-side ``ORDER BY`` (SPEC BP1): + * ``last_heard`` for nodes, ``rx_time`` for the rest. + * + * @type {ReadonlyArray} + */ + const COLLECTION_BACKFILLS = [ + { + name: 'nodes', + fetchPage: (limit, before) => fetchNodes(limit, 0, { before }), + idOf: row => row && row.node_id, + cursorOf: row => row && row.last_heard, + merge: batch => { allNodes = mergeById(allNodes, batch, 'node_id'); }, + refine: () => rebuildNodeDerivedState(), + }, + { + name: 'positions', + fetchPage: (limit, before) => fetchPositions(limit, 0, { before }), + idOf: row => row && row.id, + cursorOf: row => row && row.rx_time, + merge: batch => { + allPositionEntries = trimToWindow(mergeById(allPositionEntries, batch, 'id'), recentBackfillFloor()); + }, + refine: () => rebuildNodeDerivedState(), + }, + { + name: 'telemetry', + fetchPage: (limit, before) => fetchTelemetry(limit, 0, { before }), + idOf: row => row && row.id, + cursorOf: row => row && row.rx_time, + merge: batch => { + allTelemetryEntries = trimToWindow(mergeById(allTelemetryEntries, batch, 'id'), recentBackfillFloor()); + }, + refine: () => rebuildNodeDerivedState(), + }, + { + name: 'neighbors', + fetchPage: (limit, before) => fetchNeighbors(limit, 0, { before }), + // Composite primary key (node_id, neighbor_id) — unique per tuple, so the + // inclusive boundary row is de-duplicated across pages. + idOf: row => (row ? `${row.node_id}|${row.neighbor_id}` : undefined), + cursorOf: row => row && row.rx_time, + merge: batch => { + allNeighbors = trimToWindow( + mergeByCompositeKey(allNeighbors, batch, ['node_id', 'neighbor_id']), + longBackfillFloor(), + ); + }, + refine: () => { allNeighbors = aggregateNeighborSnapshots(allNeighbors); }, + }, + { + name: 'traces', + // Fetch raw (applyAgeFilter:false) so the pager sees the server's true page + // length for short-page termination; the age bound is applied by the + // window-trim in merge() instead. + fetchPage: (limit, before) => fetchTraces(limit, 0, { before, applyAgeFilter: false }), + idOf: row => row && row.id, + cursorOf: row => row && row.rx_time, + merge: batch => { + allTraces = trimToWindow(mergeById(allTraces, batch, 'id'), longBackfillFloor()); + }, + refine: () => {}, + }, + ]; + + /** + * Merge one streamed backward page into the module state, re-derive what it + * feeds, and repaint — progressively, one page at a time, mirroring the chat + * history backfill (issue #802). The merge + re-render is synchronous (no + * ``await``) so it cannot interleave with a concurrent refresh or another + * collection's commit. Each page is network-spaced, so the per-page render is + * not a hot loop; the stats fetch is skipped (the authoritative count is + * server-computed and unchanged by how many rows the client has paged in). + * + * @param {Object} spec One {@link COLLECTION_BACKFILLS} entry. + * @param {Array} batch Freshly-seen rows for this page (the pager only + * ever yields a non-empty array). + * @returns {void} + */ + function commitBackfillPage(spec, batch) { + spec.merge(batch); + spec.refine(); + renderFilteredOutputs(); + } + + /** + * Page one collection backward from its live frontier, committing+rendering + * each page, until the visibility window is exhausted. A no-op (no request) + * when the collection recorded no frontier — a short newest page or a warm-cache + * load. Errors are swallowed (logged) so one failing stream never aborts the + * others or the rendered newest page — mirroring {@link backfillChatHistory}. + * + * @param {Object} spec One {@link COLLECTION_BACKFILLS} entry. + * @returns {Promise} Resolves when this collection's window is exhausted. + */ + async function backfillCollection(spec) { + const before = collectionLiveFrontiers[spec.name]; + if (!(before > 0)) return; + try { + for await (const batch of paginateCollection(spec.fetchPage, { + limit: NODE_LIMIT, + before, + idOf: spec.idOf, + cursorOf: spec.cursorOf, + })) { + commitBackfillPage(spec, batch); + } + } catch (err) { + console.warn(`${spec.name} backfill failed; showing the most recent page only`, err); + } + } + + /** + * One-shot background backfill of every bulk collection (issue #832). After + * the first paint has rendered each collection's newest page, page the five + * collections backward through their visibility windows concurrently, each + * committing+repainting its pages as they arrive. Each {@link backfillCollection} + * self-gates on its frontier, so a collection with none (a short newest page, or + * a warm-cache load) returns without a request and the fan-out is a clean no-op + * when there is nothing to page — no pointless request, no empty long-load. + * Invoked once (guarded by ``collectionsBackfilled`` in {@link refresh}). + * + * @returns {Promise} Resolves when every collection's window is exhausted. + */ + async function backfillAllCollections() { + await Promise.all(COLLECTION_BACKFILLS.map(spec => backfillCollection(spec))); + } + /** * Compute distance from the configured map center. * @@ -4359,6 +4544,33 @@ export function initializeApp(config) { }); } + /** + * Render the filter-dependent outputs — node table, map markers, sort + * indicators, and chat log — from the current in-memory state, **without** the + * ``/api/stats`` fetch. {@link applyFilter} composes this with the stats + * refresh; the background collection backfill (issue #832) calls it directly so + * streaming a history page repaints the table/map without firing a redundant + * authoritative-count request per page — the ``/api/stats`` count is + * server-computed and unaffected by how many rows the client has paged in. + * + * @param {string} [filterQuery] Raw filter text for substring highlighting; + * defaults to the current filter input value. + * @returns {void} + */ + function renderFilteredOutputs(filterQuery = filterInput ? filterInput.value : '') { + // Text and role filters apply only to the node table and map; the chat log + // always receives the full node collection so reply-thread lookups succeed + // even for nodes that are currently hidden by the active filter. + const sortedNodes = getFilteredSortedNodes(); + const nowSec = Date.now() / 1000; + renderTable(sortedNodes, nowSec); + renderMap(sortedNodes, nowSec); + updateSortIndicators(); + // Pass the raw filterQuery (not the normalised form) so the chat log can + // highlight matching substrings in their original case. + rerenderChatLog(filterQuery); + } + /** * Apply text and role filters to the node list and re-render outputs. * @@ -4367,15 +4579,10 @@ export function initializeApp(config) { function applyFilter() { updateFilterClearVisibility(); const filterQuery = filterInput ? filterInput.value : ''; - // Text and role filters apply only to the node table and map; the chat log - // always receives the full node collection so reply-thread lookups succeed - // even for nodes that are currently hidden by the active filter. - const sortedNodes = getFilteredSortedNodes(); - const nowSec = Date.now()/1000; - renderTable(sortedNodes, nowSec); - renderMap(sortedNodes, nowSec); + renderFilteredOutputs(filterQuery); // Show an immediate local estimate for the title so it doesn't flicker // to (0) while waiting for the async /api/stats response. + const nowSec = Date.now() / 1000; const localStats = computeLocalActiveNodeStats(allNodes, nowSec); updateTitleCount(adjustStatsForHiddenProtocols(localStats)); // Title, legend, footer, and visibility are then corrected by /api/stats @@ -4389,10 +4596,6 @@ export function initializeApp(config) { updateFooterStats(visibleStats); applyProtocolVisibility(stats); }); - updateSortIndicators(); - // Pass the raw filterQuery (not the normalised form) so the chat log can - // highlight matching substrings in their original case. - rerenderChatLog(filterQuery); } // Re-filter on every keystroke so the table and map stay in sync with the @@ -4529,56 +4732,71 @@ export function initializeApp(config) { if (incomingNbTs > lastNeighborTimestamp) lastNeighborTimestamp = incomingNbTs; if (incomingTrTs > lastTraceTimestamp) lastTraceTimestamp = incomingTrTs; - // Merge incremental results with existing data. On first load the - // existing arrays are empty so the merge is effectively a no-op. - // Merge incremental results with existing data then trim to the - // configured limits so long-running tabs do not accumulate stale - // entries beyond what the server would return on a fresh fetch. - const nodes = useSince ? mergeById(allNodes, incomingNodes, 'node_id') : incomingNodes; - const positions = useSince - ? trimToLimit(mergeById(allPositionEntries, incomingPositions, 'id'), NODE_LIMIT) + // Capture each bulk collection's live frontier (oldest cursor of the + // newest page) so the one-shot background backfill (issue #832) can page + // backward from there, exactly like chatLiveFrontier. Cold load only: a + // warm-cache load already has the deeper history seeded (and its useSince + // delta pages are short), and a *short* newest page means the window is + // already exhausted — both record 0 so no pointless backward request fires + // (avoids an empty, long-loading page). The cursor column matches each + // route's server-side ORDER BY: last_heard for nodes, rx_time for the rest. + if (!useSince) { + const frontierIfFull = (rows, cap, keys) => + Array.isArray(rows) && rows.length >= cap ? minRecordTimestamp(rows, keys) : 0; + collectionLiveFrontiers = { + nodes: frontierIfFull(incomingNodes, NODE_LIMIT, ['last_heard']), + positions: frontierIfFull(incomingPositions, NODE_LIMIT, ['rx_time']), + telemetry: frontierIfFull(incomingTelemetry, NODE_LIMIT, ['rx_time']), + neighbors: frontierIfFull(incomingNeighbors, NODE_LIMIT, ['rx_time']), + traces: frontierIfFull(incomingTraces, TRACE_LIMIT, ['rx_time']), + }; + } + + // Merge incremental results into the module-level collections. On first + // load the existing arrays are empty so the merge is effectively a no-op. + // The per-packet collections (positions/telemetry/neighbors/traces) are + // bounded by their server visibility window rather than a fixed row count + // (issue #832): a count cap would trim a background backfill's older pages + // straight back out on the next refresh tick. Windows mirror FC3 — 7 d for + // positions/telemetry (like messages), 28 d for neighbors/traces. + const nowSeconds = Math.floor(Date.now() / 1000); + const messageWindowFloor = nowSeconds - CHAT_RECENT_WINDOW_SECONDS; + const recentWindowFloor = nowSeconds - CHAT_RECENT_WINDOW_SECONDS; + const longWindowFloor = nowSeconds - TRACE_MAX_AGE_SECONDS; + allNodes = useSince ? mergeById(allNodes, incomingNodes, 'node_id') : incomingNodes; + allPositionEntries = useSince + ? trimToWindow(mergeById(allPositionEntries, incomingPositions, 'id'), recentWindowFloor) : incomingPositions; - const neighborTuples = useSince - ? mergeByCompositeKey(allNeighbors, incomingNeighbors, ['node_id', 'neighbor_id']) - : incomingNeighbors; - const telemetryEntries = useSince - ? trimToLimit(mergeById(allTelemetryEntries, incomingTelemetry, 'id'), NODE_LIMIT) + allTelemetryEntries = useSince + ? trimToWindow(mergeById(allTelemetryEntries, incomingTelemetry, 'id'), recentWindowFloor) : incomingTelemetry; - const traceEntries = useSince - ? trimToLimit(mergeById(allTraces, incomingTraces, 'id'), TRACE_LIMIT) + allNeighbors = useSince + ? trimToWindow(mergeByCompositeKey(allNeighbors, incomingNeighbors, ['node_id', 'neighbor_id']), longWindowFloor) + : incomingNeighbors; + allTraces = useSince + ? trimToWindow(mergeById(allTraces, incomingTraces, 'id'), longWindowFloor) : incomingTraces; - // Plaintext chat is shown for the full seven-day window (issue #796), so - // bound the retained set by that window rather than a row count — a count - // cap would silently drop older-but-in-window messages on the next merge. - const messageWindowFloor = Math.floor(Date.now() / 1000) - CHAT_RECENT_WINDOW_SECONDS; - const messages = useSince - ? trimToWindow(mergeById(allMessages, incomingMessages, 'id'), messageWindowFloor) - : incomingMessages; // Encrypted blobs only feed the mixed Log tab (itself capped), so a count // cap is the right memory bound for them. const encryptedMessages = useSince ? trimToLimit(mergeById(allEncryptedMessages, incomingEncryptedMessages, 'id'), MESSAGE_LIMIT) : incomingEncryptedMessages; + // Plaintext chat is shown for the full seven-day window (issue #796), so + // bound the retained set by that window rather than a row count — a count + // cap would silently drop older-but-in-window messages on the next merge. + const messages = useSince + ? trimToWindow(mergeById(allMessages, incomingMessages, 'id'), messageWindowFloor) + : incomingMessages; - // Collapse per-source snapshot arrays into single merged records; the - // snapshot window de-duplicates entries from multiple ingestors. - const aggregatedNodes = aggregateNodeSnapshots(nodes); - const aggregatedPositions = aggregatePositionSnapshots(positions); - const aggregatedNeighbors = aggregateNeighborSnapshots(neighborTuples); - const aggregatedTelemetry = aggregateTelemetrySnapshots(telemetryEntries); - // Enrich merged node records with display name, position, distance, and - // telemetry before any rendering or filtering takes place. - aggregatedNodes.forEach(applyNodeNameFallback); - mergePositionsIntoNodes(aggregatedNodes, aggregatedPositions); - computeDistances(aggregatedNodes); - mergeTelemetryIntoNodes(aggregatedNodes, aggregatedTelemetry); - normalizeNodeCollection(aggregatedNodes); - allNodes = aggregatedNodes; - // Rebuild lookup maps after every refresh so marker updates and message - // hydration always resolve to the latest node objects. - rebuildNodeIndex(allNodes); - // Hydrate messages with node metadata in parallel; the node index must be - // rebuilt first so lookups find the freshly merged records. + // Collapse per-source snapshots and enrich the node collection from the + // merged sources. Shared with the background backfill so a streamed page + // re-derives identically (issue #832); this also re-aggregates and stores + // allPositionEntries/allTelemetryEntries/allNeighbors in their collapsed + // form (allTraces is not node-derived and keeps the merged value above). + rebuildNodeDerivedState(); + // Hydrate messages with node metadata in parallel; the node index has just + // been rebuilt (inside rebuildNodeDerivedState) so lookups find the freshly + // merged records. const [chatMessages, encryptedChatMessages] = await Promise.all([ messageNodeHydrator.hydrate(messages, nodesById), messageNodeHydrator.hydrate(encryptedMessages, nodesById) @@ -4594,10 +4812,6 @@ export function initializeApp(config) { ? trimToWindow(mergeById(allMessages, hydratedChat, 'id'), messageWindowFloor) : hydratedChat; allEncryptedMessages = Array.isArray(encryptedChatMessages) ? encryptedChatMessages : []; - allTelemetryEntries = aggregatedTelemetry; - allPositionEntries = aggregatedPositions; - allNeighbors = aggregatedNeighbors; - allTraces = Array.isArray(traceEntries) ? traceEntries : []; initialFetchDone = true; applyFilter(); // SPEC VF2/VF3/VF4: only an SSE-ping refresh flashes (refreshOptions.flash), @@ -4621,6 +4835,15 @@ export function initializeApp(config) { backfillPromise = backfillChatHistory(); void backfillPromise; } + // Likewise stream the rest of every bulk collection's window in the + // background (issue #832) so the node table / map fill past the newest + // 1000-row page the server returns at once. One-shot after the first load; + // a no-op when no collection recorded a frontier (warm load / short page). + if (!collectionsBackfilled) { + collectionsBackfilled = true; + collectionBackfillPromise = backfillAllCollections(); + void collectionBackfillPromise; + } } catch (e) { console.error(e); } @@ -4854,6 +5077,16 @@ export function initializeApp(config) { }, /** Number of plaintext chat messages currently loaded (test use only). */ getLoadedMessageCount: () => allMessages.length, + /** Number of node rows currently loaded into the table (test use only). */ + getLoadedNodeCount: () => allNodes.length, + /** Number of position entries currently loaded (test use only). */ + getLoadedPositionCount: () => allPositionEntries.length, + /** Number of telemetry entries currently loaded (test use only). */ + getLoadedTelemetryCount: () => allTelemetryEntries.length, + /** Number of neighbour tuples currently loaded (test use only). */ + getLoadedNeighborCount: () => allNeighbors.length, + /** Number of trace entries currently loaded (test use only). */ + getLoadedTraceCount: () => allTraces.length, /** The persistent data cache instance (test use only). */ dataCache, /** Seed in-memory state from the persistent cache (test use only). */ @@ -4864,6 +5097,8 @@ export function initializeApp(config) { flushCacheWrites: () => pendingCacheWrite, /** Promise resolving once the one-shot chat-history backfill finishes (test hook). */ flushBackfill: () => backfillPromise, + /** Promise resolving once the one-shot bulk-collection backfill finishes (test hook). */ + flushCollectionBackfills: () => collectionBackfillPromise, /** Empty the persistent cache — the "clear cached data" control (FC4). */ clearDataCache, /** Project an original lat/lon + pixel offset into a display LatLng. */ diff --git a/web/public/assets/js/app/main/__tests__/data-fetchers.test.js b/web/public/assets/js/app/main/__tests__/data-fetchers.test.js index bbd2843..a21fc90 100644 --- a/web/public/assets/js/app/main/__tests__/data-fetchers.test.js +++ b/web/public/assets/js/app/main/__tests__/data-fetchers.test.js @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import { fetchAllMessages, + paginateCollection, paginateMessages, fetchMessages, fetchNeighbors, @@ -526,6 +527,163 @@ test('paginateMessages stops cleanly when a page is not an array', async () => { } }); +// --------------------------------------------------------------------------- +// before cursor forwarding (issue #832 backward pagination) +// --------------------------------------------------------------------------- + +for (const [name, fn, path] of [ + ['fetchNodes', fetchNodes, '/api/nodes'], + ['fetchPositions', fetchPositions, '/api/positions'], + ['fetchTelemetry', fetchTelemetry, '/api/telemetry'], + ['fetchNeighbors', fetchNeighbors, '/api/neighbors'], + ['fetchTraces', fetchTraces, '/api/traces'], +]) { + test(`${name} forwards a positive before cursor and omits a non-positive one`, async () => { + const stub = withFetchStub({ ok: true, body: [] }); + try { + await fn(50, 0, { before: 4242 }); + assert.ok(stub.calls[0].url.startsWith(`${path}?`)); + assert.ok(stub.calls[0].url.includes('before=4242'), stub.calls[0].url); + await fn(50, 0, { before: 0 }); + assert.ok(!stub.calls[1].url.includes('before='), stub.calls[1].url); + } finally { + stub.restore(); + } + }); +} + +test('fetchTraces with applyAgeFilter:false returns rows verbatim (raw page length preserved)', async () => { + const nowSeconds = Math.floor(Date.now() / 1000); + const stub = withFetchStub({ + ok: true, + body: [ + { id: 1, rx_time: nowSeconds }, + { id: 2, rx_time: nowSeconds - 365 * 24 * 3600 }, // expired — kept when the filter is off + ], + }); + try { + const raw = await fetchTraces(50, 0, { applyAgeFilter: false }); + assert.deepEqual(raw.map(t => t.id), [1, 2]); + } finally { + stub.restore(); + } +}); + +// --------------------------------------------------------------------------- +// paginateCollection (issue #832 generic backward pager) +// --------------------------------------------------------------------------- + +test('paginateCollection pages backward by cursor and de-duplicates by id', async () => { + // limit=2; the inclusive cursor re-returns the boundary row, which must be + // de-duplicated out of the later page. + const pages = { + 0: [{ id: 5, t: 50 }, { id: 4, t: 40 }], + 40: [{ id: 4, t: 40 }, { id: 3, t: 30 }], + 30: [{ id: 3, t: 30 }], // short → stop (all seen) + }; + const seenBefore = []; + const fetchPage = async (limit, before) => { + seenBefore.push(before); + return pages[before] || pages[0]; + }; + const out = []; + for await (const batch of paginateCollection(fetchPage, { limit: 2, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch.map(r => r.id)); + } + assert.deepEqual(out, [[5, 4], [3]]); + assert.deepEqual(seenBefore, [0, 40, 30]); +}); + +test('paginateCollection seeds the first request from a positive before and stops on a short page', async () => { + const calls = []; + const fetchPage = async (limit, before) => { + calls.push(before); + return before === 500 ? [{ id: 9, t: 90 }] : []; + }; + const out = []; + for await (const batch of paginateCollection(fetchPage, { limit: 2, before: 500, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch.map(r => r.id)); + } + assert.deepEqual(out, [[9]]); + assert.deepEqual(calls, [500]); +}); + +test('paginateCollection ignores a non-finite before seed', async () => { + const calls = []; + const fetchPage = async (limit, before) => { + calls.push(before); + return [{ id: 1, t: 10 }]; + }; + const out = []; + for await (const batch of paginateCollection(fetchPage, { limit: 2, before: Number.NaN, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch); + } + assert.equal(calls[0], 0); +}); + +test('paginateCollection makes one call and yields nothing for an empty window', async () => { + let calls = 0; + const out = []; + for await (const batch of paginateCollection(async () => { calls += 1; return []; }, { limit: 2, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch); + } + assert.deepEqual(out, []); + assert.equal(calls, 1); +}); + +test('paginateCollection stops on a no-progress page (server ignored the cursor)', async () => { + // The same full page is returned regardless of the cursor; without the + // no-progress guard this would loop forever. + const out = []; + for await (const batch of paginateCollection(async () => [{ id: 5, t: 50 }, { id: 4, t: 40 }], { limit: 2, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch.map(r => r.id)); + } + assert.deepEqual(out, [[5, 4]]); +}); + +test('paginateCollection stops when no row carries a usable cursor', async () => { + const calls = []; + const fetchPage = async (limit, before) => { + calls.push(before); + return [{ id: 7 }, { id: 8 }]; // full page, no cursor field + }; + const out = []; + for await (const batch of paginateCollection(fetchPage, { limit: 2, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch.map(r => r.id)); + } + assert.deepEqual(out, [[7, 8]]); + assert.equal(calls.length, 1); // oldest stayed 0 → stop after one page +}); + +test('paginateCollection skips rows without an id and then stops (no progress)', async () => { + let calls = 0; + const out = []; + for await (const batch of paginateCollection(async () => { calls += 1; return [{ t: 40 }, { t: 30 }]; }, { limit: 2, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch); + } + assert.deepEqual(out, []); + assert.equal(calls, 1); +}); + +test('paginateCollection honours the maxPages backstop against a runaway feed', async () => { + let n = 0; + const fetchPage = async () => { n += 1; return [{ id: 100 - n, t: 100 - n }]; }; // always full, strictly older + let pages = 0; + // eslint-disable-next-line no-unused-vars + for await (const _ of paginateCollection(fetchPage, { limit: 1, maxPages: 3, idOf: r => r.id, cursorOf: r => r.t })) { + pages += 1; + } + assert.equal(pages, 3); +}); + +test('paginateCollection stops cleanly when a page is not an array', async () => { + const out = []; + for await (const batch of paginateCollection(async () => ({ error: 'nope' }), { limit: 2, idOf: r => r.id, cursorOf: r => r.t })) { + out.push(batch); + } + assert.deepEqual(out, []); +}); + // --------------------------------------------------------------------------- // responsePromise (cold-load boot prefetch consumption) // --------------------------------------------------------------------------- diff --git a/web/public/assets/js/app/main/data-fetchers.js b/web/public/assets/js/app/main/data-fetchers.js index 43ddfd7..6595508 100644 --- a/web/public/assets/js/app/main/data-fetchers.js +++ b/web/public/assets/js/app/main/data-fetchers.js @@ -89,14 +89,17 @@ function resolveResponse(responsePromise, url) { * * @param {number} [limit=NODE_LIMIT] Maximum number of records. * @param {number} [since=0] Unix timestamp; only rows newer than this are returned. - * @param {{ responsePromise?: Promise }} [options] Optional pre-issued - * boot-prefetch response to consume instead of issuing a fresh request. + * @param {{ responsePromise?: Promise, before?: number }} [options] + * Optional pre-issued boot-prefetch response to consume instead of issuing a + * fresh request, and an inclusive upper-bound ``last_heard`` cursor for + * backward pagination (SPEC BP1; issue #832). * @returns {Promise>} Parsed node payloads. */ -export async function fetchNodes(limit = NODE_LIMIT, since = 0, { responsePromise } = {}) { +export async function fetchNodes(limit = NODE_LIMIT, since = 0, { responsePromise, before = 0 } = {}) { const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT); let url = `/api/nodes?limit=${effectiveLimit}`; if (since > 0) url += `&since=${since}`; + if (before > 0) url += `&before=${before}`; const r = await resolveResponse(responsePromise, url); if (!r.ok) throw new Error('HTTP ' + r.status); return r.json(); @@ -153,46 +156,48 @@ export async function fetchMessages(limit, options = {}) { } /** - * Page backward through the message feed, ``yield``-ing each page's new rows. + * Generic backward pager shared by every bulk collection (issue #832; the + * deferred SPEC BP9a follow-up). * - * The API clamps each response to {@link MESSAGE_LIMIT} rows ordered newest → - * oldest, so the whole window is only reachable by walking an inclusive - * ``before`` cursor: pull a page, then re-request everything at or before the - * oldest ``rx_time`` seen. Rows are de-duplicated by ``id`` across pages (the - * inclusive cursor re-returns each boundary row), and each page's freshly-seen - * rows are yielded as soon as they arrive so callers can render progressively - * instead of waiting for the entire window (issue #802). + * The API clamps each response to its per-route cap ordered newest → oldest, so + * the whole window is only reachable by walking an inclusive ``before`` cursor: + * pull a page, then re-request everything at or before the oldest cursor value + * seen. Rows are de-duplicated by ``idOf`` across pages (the inclusive cursor + * re-returns each boundary row), and each page's freshly-seen rows are yielded as + * soon as they arrive so callers render progressively instead of awaiting the + * whole window. * * Iteration stops when a short page signals the window is exhausted, when a page * yields no new rows (the server ignored the cursor, or every row was a boundary - * duplicate), when no row carries a usable timestamp cursor, or when ``maxPages`` - * is hit as a runaway backstop. + * duplicate), when no row carries a usable cursor value, or when ``maxPages`` is + * hit as a runaway backstop. * - * @param {number} limit Page size (rows requested per API call). - * @param {{ encrypted?: boolean, before?: number, chatEnabled?: boolean, normaliseMessageLimit?: Function, maxPages?: number }} [options] - * Retrieval flags, dependency hooks, an optional ``before`` cursor to seed the - * walk from (used to resume paging just past an already-loaded newest page), - * and an optional page-count backstop. + * @param {(limit: number, before: number) => Promise>} fetchPage + * Fetches one page bounded by an inclusive ``before`` cursor (0 ⇒ newest page). + * @param {{ limit: number, before?: number, maxPages?: number, + * idOf: (row: Object) => *, cursorOf: (row: Object) => * }} options + * ``limit`` page size; ``before`` optional seed cursor (resume just past an + * already-loaded newest page); ``maxPages`` runaway backstop; ``idOf`` per-row + * dedup key; ``cursorOf`` per-row cursor value (the column the route orders by). * @yields {Array} The de-duplicated new rows of each page. * @returns {AsyncGenerator>} */ -export async function* paginateMessages(limit, options = {}) { - const { maxPages = 200, before: initialBefore = 0, ...fetchOptions } = options; +export async function* paginateCollection(fetchPage, { limit, before: initialBefore = 0, maxPages = 200, idOf, cursorOf }) { const seen = new Set(); let before = Number.isFinite(initialBefore) && initialBefore > 0 ? initialBefore : 0; for (let page = 0; page < maxPages; page += 1) { // eslint-disable-next-line no-await-in-loop -- pages are inherently sequential (cursor depends on the prior page). - const batch = await fetchMessages(limit, { ...fetchOptions, before }); + const batch = await fetchPage(limit, before); if (!Array.isArray(batch) || batch.length === 0) return; const fresh = []; let oldest = 0; - for (const message of batch) { - const id = message && message.id; + for (const row of batch) { + const id = idOf(row); if (id != null && !seen.has(id)) { seen.add(id); - fresh.push(message); + fresh.push(row); } - const ts = Number(message && message.rx_time); + const ts = Number(cursorOf(row)); if (Number.isFinite(ts) && (oldest === 0 || ts < oldest)) { oldest = ts; } @@ -207,6 +212,37 @@ export async function* paginateMessages(limit, options = {}) { } } +/** + * Page backward through the message feed, ``yield``-ing each page's new rows. + * + * A thin specialisation of {@link paginateCollection} for the chat feed: rows + * are keyed by ``id`` and the cursor is ``rx_time`` (the column the messages + * route orders by). Each page's freshly-seen rows are yielded as soon as they + * arrive so callers can render progressively instead of waiting for the entire + * window (issue #802). + * + * @param {number} limit Page size (rows requested per API call). + * @param {{ encrypted?: boolean, before?: number, chatEnabled?: boolean, normaliseMessageLimit?: Function, maxPages?: number }} [options] + * Retrieval flags, dependency hooks, an optional ``before`` cursor to seed the + * walk from (used to resume paging just past an already-loaded newest page), + * and an optional page-count backstop. + * @yields {Array} The de-duplicated new rows of each page. + * @returns {AsyncGenerator>} + */ +export async function* paginateMessages(limit, options = {}) { + const { maxPages = 200, before = 0, ...fetchOptions } = options; + yield* paginateCollection( + (pageLimit, pageBefore) => fetchMessages(pageLimit, { ...fetchOptions, before: pageBefore }), + { + limit, + before, + maxPages, + idOf: row => row && row.id, + cursorOf: row => row && row.rx_time, + }, + ); +} + /** * Eagerly page *every* message in the server's visibility window into one array. * @@ -233,14 +269,17 @@ export async function fetchAllMessages(limit, options = {}) { * * @param {number} [limit=NODE_LIMIT] Maximum number of rows. * @param {number} [since=0] Unix timestamp; only rows newer than this are returned. - * @param {{ responsePromise?: Promise }} [options] Optional pre-issued - * boot-prefetch response to consume instead of issuing a fresh request. + * @param {{ responsePromise?: Promise, before?: number }} [options] + * Optional pre-issued boot-prefetch response to consume instead of issuing a + * fresh request, and an inclusive upper-bound ``rx_time`` cursor for backward + * pagination (SPEC BP1; issue #832). * @returns {Promise>} Parsed neighbour payloads. */ -export async function fetchNeighbors(limit = NODE_LIMIT, since = 0, { responsePromise } = {}) { +export async function fetchNeighbors(limit = NODE_LIMIT, since = 0, { responsePromise, before = 0 } = {}) { const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT); let url = `/api/neighbors?limit=${effectiveLimit}`; if (since > 0) url += `&since=${since}`; + if (before > 0) url += `&before=${before}`; const r = await resolveResponse(responsePromise, url); if (!r.ok) throw new Error('HTTP ' + r.status); return r.json(); @@ -251,19 +290,25 @@ export async function fetchNeighbors(limit = NODE_LIMIT, since = 0, { responsePr * * @param {number} [limit=TRACE_LIMIT] Maximum number of records. * @param {number} [since=0] Unix timestamp; only rows newer than this are returned. - * @param {{ responsePromise?: Promise }} [options] Optional pre-issued - * boot-prefetch response to consume instead of issuing a fresh request. + * @param {{ responsePromise?: Promise, before?: number, applyAgeFilter?: boolean }} [options] + * Optional pre-issued boot-prefetch response to consume instead of issuing a + * fresh request; an inclusive upper-bound ``rx_time`` cursor for backward + * pagination (SPEC BP1; issue #832); and ``applyAgeFilter`` (default ``true``) + * which, when ``false``, returns the server's rows verbatim so a backward pager + * sees the true page length for its short-page termination (the client age + * filter is re-applied at merge time instead). * @returns {Promise>} Parsed trace payloads. */ -export async function fetchTraces(limit = TRACE_LIMIT, since = 0, { responsePromise } = {}) { +export async function fetchTraces(limit = TRACE_LIMIT, since = 0, { responsePromise, before = 0, applyAgeFilter = true } = {}) { const safeLimit = Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : TRACE_LIMIT; const effectiveLimit = Math.min(safeLimit, NODE_LIMIT); let url = `/api/traces?limit=${effectiveLimit}`; if (since > 0) url += `&since=${since}`; + if (before > 0) url += `&before=${before}`; const r = await resolveResponse(responsePromise, url); if (!r.ok) throw new Error('HTTP ' + r.status); const traces = await r.json(); - return filterRecentTraces(traces, TRACE_MAX_AGE_SECONDS); + return applyAgeFilter ? filterRecentTraces(traces, TRACE_MAX_AGE_SECONDS) : traces; } /** @@ -271,14 +316,17 @@ export async function fetchTraces(limit = TRACE_LIMIT, since = 0, { responseProm * * @param {number} [limit=NODE_LIMIT] Maximum number of rows. * @param {number} [since=0] Unix timestamp; only rows newer than this are returned. - * @param {{ responsePromise?: Promise }} [options] Optional pre-issued - * boot-prefetch response to consume instead of issuing a fresh request. + * @param {{ responsePromise?: Promise, before?: number }} [options] + * Optional pre-issued boot-prefetch response to consume instead of issuing a + * fresh request, and an inclusive upper-bound ``rx_time`` cursor for backward + * pagination (SPEC BP1; issue #832). * @returns {Promise>} Parsed telemetry payloads. */ -export async function fetchTelemetry(limit = NODE_LIMIT, since = 0, { responsePromise } = {}) { +export async function fetchTelemetry(limit = NODE_LIMIT, since = 0, { responsePromise, before = 0 } = {}) { const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT); let url = `/api/telemetry?limit=${effectiveLimit}`; if (since > 0) url += `&since=${since}`; + if (before > 0) url += `&before=${before}`; const r = await resolveResponse(responsePromise, url); if (!r.ok) throw new Error('HTTP ' + r.status); return r.json(); @@ -289,14 +337,17 @@ export async function fetchTelemetry(limit = NODE_LIMIT, since = 0, { responsePr * * @param {number} [limit=NODE_LIMIT] Maximum number of rows. * @param {number} [since=0] Unix timestamp; only rows newer than this are returned. - * @param {{ responsePromise?: Promise }} [options] Optional pre-issued - * boot-prefetch response to consume instead of issuing a fresh request. + * @param {{ responsePromise?: Promise, before?: number }} [options] + * Optional pre-issued boot-prefetch response to consume instead of issuing a + * fresh request, and an inclusive upper-bound ``rx_time`` cursor for backward + * pagination (SPEC BP1; issue #832). * @returns {Promise>} Parsed position payloads. */ -export async function fetchPositions(limit = NODE_LIMIT, since = 0, { responsePromise } = {}) { +export async function fetchPositions(limit = NODE_LIMIT, since = 0, { responsePromise, before = 0 } = {}) { const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT); let url = `/api/positions?limit=${effectiveLimit}`; if (since > 0) url += `&since=${since}`; + if (before > 0) url += `&before=${before}`; const r = await resolveResponse(responsePromise, url); if (!r.ok) throw new Error('HTTP ' + r.status); return r.json();