From 5b0a6f5f8b1f6c44ed494e8f5ced10636e321fca Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sat, 14 Feb 2026 21:14:10 +0100 Subject: [PATCH] web: expose node stats in distinct api (#641) * web: expose node stats in distinct api * web: address review comments * web: address review comments * web: address review comments * web: address review comments --- web/lib/potato_mesh/application/federation.rb | 58 ++++- web/lib/potato_mesh/application/queries.rb | 37 +++ web/lib/potato_mesh/application/routes/api.rb | 8 + .../js/app/__tests__/main-stats.test.js | 210 +++++++++++++++++ web/public/assets/js/app/main.js | 160 ++++++++++++- web/spec/app_spec.rb | 42 ++++ web/spec/federation_spec.rb | 218 +++++++++++++----- 7 files changed, 652 insertions(+), 81 deletions(-) create mode 100644 web/public/assets/js/app/__tests__/main-stats.test.js diff --git a/web/lib/potato_mesh/application/federation.rb b/web/lib/potato_mesh/application/federation.rb index 78b9c15..6143710 100644 --- a/web/lib/potato_mesh/application/federation.rb +++ b/web/lib/potato_mesh/application/federation.rb @@ -739,6 +739,34 @@ module PotatoMesh [nil, errors] end + # Resolve the best matching active-node count from a remote /api/stats payload. + # + # @param payload [Hash, nil] decoded JSON payload from /api/stats. + # @param max_age_seconds [Integer] activity window currently expected for federation freshness. + # @return [Integer, nil] selected active-node count when available. + def remote_active_node_count_from_stats(payload, max_age_seconds:) + return nil unless payload.is_a?(Hash) + + active_nodes = payload["active_nodes"] + return nil unless active_nodes.is_a?(Hash) + + age = coerce_integer(max_age_seconds) || 0 + key = if age <= 3600 + "hour" + elsif age <= 86_400 + "day" + elsif age <= PotatoMesh::Config.week_seconds + "week" + else + "month" + end + + value = coerce_integer(active_nodes[key]) + return nil unless value + + [value, 0].max + end + # Parse a remote federation instance payload into canonical attributes. # # @param payload [Hash] JSON object describing a remote instance. @@ -1049,21 +1077,33 @@ module PotatoMesh attributes[:is_private] = false if attributes[:is_private].nil? + stats_payload, stats_metadata = fetch_instance_json(attributes[:domain], "/api/stats") + stats_count = remote_active_node_count_from_stats( + stats_payload, + max_age_seconds: PotatoMesh::Config.remote_instance_max_node_age, + ) + attributes[:nodes_count] = stats_count if stats_count + nodes_since_path = "/api/nodes?since=#{recent_cutoff}&limit=1000" nodes_since_window, nodes_since_metadata = fetch_instance_json(attributes[:domain], nodes_since_path) - if nodes_since_window.is_a?(Array) + if stats_count.nil? && attributes[:nodes_count].nil? && nodes_since_window.is_a?(Array) attributes[:nodes_count] = nodes_since_window.length - elsif nodes_since_metadata - warn_log( - "Failed to load remote node window", - context: "federation.instances", - domain: attributes[:domain], - reason: Array(nodes_since_metadata).map(&:to_s).join("; "), - ) end remote_nodes, node_metadata = fetch_instance_json(attributes[:domain], "/api/nodes") - remote_nodes ||= nodes_since_window if nodes_since_window.is_a?(Array) + remote_nodes = nodes_since_window if remote_nodes.nil? && nodes_since_window.is_a?(Array) + if attributes[:nodes_count].nil? && remote_nodes.is_a?(Array) + attributes[:nodes_count] = remote_nodes.length + end + + if stats_count.nil? && Array(stats_metadata).any? + debug_log( + "Remote instance /api/stats unavailable; using node list fallback", + context: "federation.instances", + domain: attributes[:domain], + reason: Array(stats_metadata).map(&:to_s).join("; "), + ) + end unless remote_nodes warn_log( "Failed to load remote node data", diff --git a/web/lib/potato_mesh/application/queries.rb b/web/lib/potato_mesh/application/queries.rb index f400246..9f969a2 100644 --- a/web/lib/potato_mesh/application/queries.rb +++ b/web/lib/potato_mesh/application/queries.rb @@ -127,6 +127,43 @@ module PotatoMesh [threshold, floor].max end + # Return exact active-node counts across common activity windows. + # + # Counts are resolved directly in SQL with COUNT(*) thresholds against + # +nodes.last_heard+ to avoid sampling bias from list endpoint limits. + # + # @param now [Integer] reference unix timestamp in seconds. + # @param db [SQLite3::Database, nil] optional open database handle to reuse. + # @return [Hash{String => Integer}] counts keyed by hour/day/week/month. + def query_active_node_stats(now: Time.now.to_i, db: nil) + handle = db || open_database(readonly: true) + handle.results_as_hash = true + reference_now = coerce_integer(now) || Time.now.to_i + hour_cutoff = reference_now - 3600 + day_cutoff = reference_now - 86_400 + week_cutoff = reference_now - PotatoMesh::Config.week_seconds + month_cutoff = reference_now - (30 * 24 * 60 * 60) + private_filter = private_mode? ? " AND (role IS NULL OR role <> 'CLIENT_HIDDEN')" : "" + sql = <<~SQL + SELECT + (SELECT COUNT(*) FROM nodes WHERE last_heard >= ?#{private_filter}) AS hour_count, + (SELECT COUNT(*) FROM nodes WHERE last_heard >= ?#{private_filter}) AS day_count, + (SELECT COUNT(*) FROM nodes WHERE last_heard >= ?#{private_filter}) AS week_count, + (SELECT COUNT(*) FROM nodes WHERE last_heard >= ?#{private_filter}) AS month_count + SQL + row = with_busy_retry do + handle.get_first_row(sql, [hour_cutoff, day_cutoff, week_cutoff, month_cutoff]) + end || {} + { + "hour" => row["hour_count"].to_i, + "day" => row["day_count"].to_i, + "week" => row["week_count"].to_i, + "month" => row["month_count"].to_i, + } + ensure + handle&.close unless db + end + def node_reference_tokens(node_ref) parts = canonical_node_parts(node_ref) canonical_id, numeric_id = parts ? parts[0, 2] : [nil, nil] diff --git a/web/lib/potato_mesh/application/routes/api.rb b/web/lib/potato_mesh/application/routes/api.rb index 5d15494..f3fd5f9 100644 --- a/web/lib/potato_mesh/application/routes/api.rb +++ b/web/lib/potato_mesh/application/routes/api.rb @@ -67,6 +67,14 @@ module PotatoMesh query_nodes(limit, since: params["since"]).to_json end + app.get "/api/stats" do + content_type :json + { + active_nodes: query_active_node_stats, + sampled: false, + }.to_json + end + app.get "/api/nodes/:id" do content_type :json node_ref = string_or_nil(params["id"]) diff --git a/web/public/assets/js/app/__tests__/main-stats.test.js b/web/public/assets/js/app/__tests__/main-stats.test.js new file mode 100644 index 0000000..954db35 --- /dev/null +++ b/web/public/assets/js/app/__tests__/main-stats.test.js @@ -0,0 +1,210 @@ +/* + * 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. + */ + +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { + computeLocalActiveNodeStats, + fetchActiveNodeStats, + formatActiveNodeStatsText, + normaliseActiveNodeStatsPayload, +} from '../main.js'; + +const NOW = 1_700_000_000; + +test('computeLocalActiveNodeStats calculates local hour/day/week/month counts', () => { + const nodes = [ + { last_heard: NOW - 60 }, + { last_heard: NOW - 4_000 }, + { last_heard: NOW - 90_000 }, + { last_heard: NOW - (8 * 86_400) }, + { last_heard: NOW - (20 * 86_400) }, + ]; + + const stats = computeLocalActiveNodeStats(nodes, NOW); + + assert.deepEqual(stats, { + hour: 1, + day: 2, + week: 3, + month: 5, + sampled: true, + }); +}); + +test('normaliseActiveNodeStatsPayload validates and normalizes API payload', () => { + const payload = { + active_nodes: { + hour: '11', + day: 22, + week: 33, + month: 44, + }, + sampled: false, + }; + + assert.deepEqual(normaliseActiveNodeStatsPayload(payload), { + hour: 11, + day: 22, + week: 33, + month: 44, + sampled: false, + }); + + assert.equal(normaliseActiveNodeStatsPayload({}), null); +}); + +test('normaliseActiveNodeStatsPayload rejects malformed stat values', () => { + assert.equal( + normaliseActiveNodeStatsPayload({ active_nodes: { hour: 'x', day: 1, week: 1, month: 1 } }), + null + ); + assert.equal( + normaliseActiveNodeStatsPayload({ active_nodes: null }), + null + ); +}); + +test('normaliseActiveNodeStatsPayload clamps negatives and truncates floats', () => { + assert.deepEqual( + normaliseActiveNodeStatsPayload({ + active_nodes: { hour: -1.9, day: 2.8, week: 3.1, month: 4.9 }, + sampled: 1 + }), + { hour: 0, day: 2, week: 3, month: 4, sampled: true } + ); +}); + +test('fetchActiveNodeStats uses /api/stats when available', async () => { + const calls = []; + const fetchImpl = async (url) => { + calls.push(url); + return { + ok: true, + async json() { + return { + active_nodes: { hour: 5, day: 15, week: 25, month: 35 }, + sampled: false, + }; + }, + }; + }; + + const stats = await fetchActiveNodeStats({ nodes: [], nowSeconds: NOW, fetchImpl }); + + assert.equal(calls[0], '/api/stats'); + assert.deepEqual(stats, { + hour: 5, + day: 15, + week: 25, + month: 35, + sampled: false, + }); +}); + +test('fetchActiveNodeStats reuses cached /api/stats response for repeated calls', async () => { + const calls = []; + const fetchImpl = async (url) => { + calls.push(url); + return { + ok: true, + async json() { + return { + active_nodes: { hour: 2, day: 4, week: 6, month: 8 }, + sampled: false, + }; + }, + }; + }; + + const first = await fetchActiveNodeStats({ nodes: [], nowSeconds: NOW, fetchImpl }); + const second = await fetchActiveNodeStats({ nodes: [], nowSeconds: NOW, fetchImpl }); + + assert.equal(calls.length, 1); + assert.deepEqual(first, second); +}); + +test('fetchActiveNodeStats falls back to local counts when stats fetch fails', async () => { + const nodes = [ + { last_heard: NOW - 120 }, + { last_heard: NOW - (10 * 86_400) }, + ]; + const fetchImpl = async () => { + throw new Error('network down'); + }; + + const stats = await fetchActiveNodeStats({ nodes, nowSeconds: NOW, fetchImpl }); + + assert.deepEqual(stats, { + hour: 1, + day: 1, + week: 1, + month: 2, + sampled: true, + }); +}); + +test('fetchActiveNodeStats falls back to local counts on non-OK HTTP responses', async () => { + const stats = await fetchActiveNodeStats({ + nodes: [{ last_heard: NOW - 10 }], + nowSeconds: NOW, + fetchImpl: async () => ({ ok: false, status: 503 }) + }); + assert.equal(stats.sampled, true); + assert.equal(stats.hour, 1); +}); + +test('fetchActiveNodeStats falls back to local counts on invalid payloads', async () => { + const stats = await fetchActiveNodeStats({ + nodes: [{ last_heard: NOW - (31 * 86_400) }], + nowSeconds: NOW, + fetchImpl: async () => ({ + ok: true, + async json() { + return { active_nodes: { hour: 'bad' } }; + } + }) + }); + assert.equal(stats.sampled, true); + assert.equal(stats.month, 0); +}); + +test('formatActiveNodeStatsText emits expected dashboard string', () => { + const text = formatActiveNodeStatsText({ + channel: 'LongFast', + frequency: '868MHz', + stats: { hour: 1, day: 2, week: 3, month: 4, sampled: false }, + }); + + assert.equal( + text, + 'LongFast (868MHz) — active nodes: 1/hour, 2/day, 3/week, 4/month.' + ); +}); + +test('formatActiveNodeStatsText appends sampled marker when local fallback is used', () => { + const text = formatActiveNodeStatsText({ + channel: 'LongFast', + frequency: '868MHz', + stats: { hour: 9, day: 8, week: 7, month: 6, sampled: true }, + }); + + assert.equal( + text, + 'LongFast (868MHz) — active nodes: 9/hour, 8/day, 7/week, 6/month (sampled).' + ); +}); diff --git a/web/public/assets/js/app/main.js b/web/public/assets/js/app/main.js index c071e04..2dedb66 100644 --- a/web/public/assets/js/app/main.js +++ b/web/public/assets/js/app/main.js @@ -69,6 +69,144 @@ import { roleRenderOrder, } from './role-helpers.js'; +/** + * Compute active-node counts from a local node array. + * + * @param {Array} nodes Node payloads. + * @param {number} nowSeconds Reference timestamp. + * @returns {{hour: number, day: number, week: number, month: number, sampled: boolean}} Local count snapshot. + */ +export function computeLocalActiveNodeStats(nodes, nowSeconds) { + const safeNodes = Array.isArray(nodes) ? nodes : []; + const referenceNow = Number.isFinite(nowSeconds) ? nowSeconds : Date.now() / 1000; + const windows = [ + { key: 'hour', secs: 3600 }, + { key: 'day', secs: 86_400 }, + { key: 'week', secs: 7 * 86_400 }, + { key: 'month', secs: 30 * 86_400 } + ]; + const counts = { sampled: true }; + for (const window of windows) { + counts[window.key] = safeNodes.filter(node => { + const lastHeard = Number(node?.last_heard); + return Number.isFinite(lastHeard) && referenceNow - lastHeard <= window.secs; + }).length; + } + return counts; +} + +/** + * Parse and validate the `/api/stats` payload. + * + * @param {*} payload Candidate JSON object from the stats endpoint. + * @returns {{hour: number, day: number, week: number, month: number, sampled: boolean}|null} Normalized stats or null. + */ +export function normaliseActiveNodeStatsPayload(payload) { + const activeNodes = payload && typeof payload === 'object' ? payload.active_nodes : null; + if (!activeNodes || typeof activeNodes !== 'object') { + return null; + } + const hour = Number(activeNodes.hour); + const day = Number(activeNodes.day); + const week = Number(activeNodes.week); + const month = Number(activeNodes.month); + if (![hour, day, week, month].every(Number.isFinite)) { + return null; + } + return { + hour: Math.max(0, Math.trunc(hour)), + day: Math.max(0, Math.trunc(day)), + week: Math.max(0, Math.trunc(week)), + month: Math.max(0, Math.trunc(month)), + sampled: Boolean(payload.sampled) + }; +} + +const ACTIVE_NODE_STATS_CACHE_TTL_MS = 30_000; +let activeNodeStatsCache = null; +let activeNodeStatsFetchPromise = null; +let activeNodeStatsFetchImpl = null; + +/** + * Fetch active-node stats from the dedicated API endpoint with short-lived caching. + * + * @param {Function} fetchImpl Fetch implementation. + * @returns {Promise<{hour: number, day: number, week: number, month: number, sampled: boolean} | null>} Normalized stats or null. + */ +async function fetchRemoteActiveNodeStats(fetchImpl) { + const nowMs = Date.now(); + if (activeNodeStatsCache?.fetchImpl === fetchImpl && activeNodeStatsCache.expiresAt > nowMs) { + return activeNodeStatsCache.stats; + } + if (activeNodeStatsFetchPromise && activeNodeStatsFetchImpl === fetchImpl) { + return activeNodeStatsFetchPromise; + } + + activeNodeStatsFetchImpl = fetchImpl; + activeNodeStatsFetchPromise = (async () => { + const response = await fetchImpl('/api/stats', { cache: 'no-store' }); + if (!response?.ok) { + throw new Error(`stats HTTP ${response?.status ?? 'unknown'}`); + } + const payload = await response.json(); + const normalized = normaliseActiveNodeStatsPayload(payload); + if (!normalized) { + throw new Error('invalid stats payload'); + } + activeNodeStatsCache = { + fetchImpl, + expiresAt: Date.now() + ACTIVE_NODE_STATS_CACHE_TTL_MS, + stats: normalized + }; + return normalized; + })(); + + try { + return await activeNodeStatsFetchPromise; + } finally { + activeNodeStatsFetchPromise = null; + activeNodeStatsFetchImpl = null; + } +} + +/** + * Fetch active-node stats from the dedicated API endpoint with local fallback. + * + * @param {{ + * nodes: Array, + * nowSeconds: number, + * fetchImpl?: Function + * }} params Fetch parameters. + * @returns {Promise<{hour: number, day: number, week: number, month: number, sampled: boolean}>} Stats snapshot. + */ +export async function fetchActiveNodeStats({ nodes, nowSeconds, fetchImpl = fetch }) { + try { + const normalized = await fetchRemoteActiveNodeStats(fetchImpl); + if (normalized) return normalized; + throw new Error('invalid stats payload'); + } catch (error) { + console.debug('Failed to fetch /api/stats; using local active-node counts.', error); + return computeLocalActiveNodeStats(nodes, nowSeconds); + } +} + +/** + * Format the dashboard refresh-info sentence for active-node counts. + * + * @param {{channel: string, frequency: string, stats: {hour:number,day:number,week:number,month:number,sampled:boolean}}} params Formatting data. + * @returns {string} User-visible sentence for the dashboard header. + */ +export function formatActiveNodeStatsText({ channel, frequency, stats }) { + const parts = [ + `${Number(stats?.hour) || 0}/hour`, + `${Number(stats?.day) || 0}/day`, + `${Number(stats?.week) || 0}/week`, + `${Number(stats?.month) || 0}/month` + ]; + const suffix = stats?.sampled ? ' (sampled)' : ''; + return `${channel} (${frequency}) — active nodes: ${parts.join(', ')}${suffix}.`; +} + /** * Entry point for the interactive dashboard. Wires up event listeners, * initializes the map, and triggers the first data refresh cycle. @@ -222,6 +360,7 @@ export function initializeApp(config) { /** @type {ReturnType|null} */ let refreshTimer = null; + let refreshInfoRequestId = 0; /** * Close any open short-info overlays that do not contain the provided anchor. @@ -4395,15 +4534,16 @@ export function initializeApp(config) { if (!refreshInfo || !isDashboardView) { return; } - const windows = [ - { label: 'hour', secs: 3600 }, - { label: 'day', secs: 86400 }, - { label: 'week', secs: 7 * 86400 }, - ]; - const counts = windows.map(w => { - const c = nodes.filter(n => n.last_heard && nowSec - Number(n.last_heard) <= w.secs).length; - return `${c}/${w.label}`; - }).join(', '); - refreshInfo.textContent = `${config.channel} (${config.frequency}) — active nodes: ${counts}.`; + const requestId = ++refreshInfoRequestId; + void fetchActiveNodeStats({ nodes, nowSeconds: nowSec }).then(stats => { + if (requestId !== refreshInfoRequestId) { + return; + } + refreshInfo.textContent = formatActiveNodeStatsText({ + channel: config.channel, + frequency: config.frequency, + stats + }); + }); } } diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 4f80ac8..5aecd7f 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -27,6 +27,7 @@ RSpec.describe "Potato Mesh Sinatra app" do let(:app) { Sinatra::Application } let(:application_class) { PotatoMesh::Application } INSERT_NODE_WITH_LAST_HEARD_SQL = "INSERT INTO nodes(node_id, num, last_heard, first_heard) VALUES (?,?,?,?)".freeze + INSERT_NODE_WITH_METADATA_SQL = "INSERT INTO nodes(node_id, num, short_name, long_name, hw_model, role, last_heard, first_heard) VALUES(?,?,?,?,?,?,?,?)".freeze SELECT_NODE_LAST_HEARD_SQL = "SELECT last_heard FROM nodes WHERE node_id = ?".freeze describe "configuration" do @@ -5664,6 +5665,47 @@ RSpec.describe "Potato Mesh Sinatra app" do end end + describe "GET /api/stats" do + it "returns exact SQL-backed activity counts without list-endpoint sampling" do + clear_database + now = reference_time.to_i + allow(Time).to receive(:now).and_return(reference_time) + + with_db do |db| + db.transaction + 1005.times do |index| + heard = now - (index % 1800) + node_id = format("!%08x", index + 1) + db.execute( + INSERT_NODE_WITH_METADATA_SQL, + [node_id, index + 1, "n#{index}", "Node #{index}", "TBEAM", "CLIENT", heard, heard], + ) + end + db.execute( + INSERT_NODE_WITH_METADATA_SQL, + ["!week0001", 200_001, "week", "Week Node", "TBEAM", "CLIENT", now - (2 * 86_400), now - (2 * 86_400)], + ) + db.execute( + INSERT_NODE_WITH_METADATA_SQL, + ["!month001", 200_002, "month", "Month Node", "TBEAM", "CLIENT", now - (20 * 86_400), now - (20 * 86_400)], + ) + db.commit + end + + get "/api/stats" + + expect(last_response).to be_ok + payload = JSON.parse(last_response.body) + expect(payload["sampled"]).to eq(false) + expect(payload["active_nodes"]).to include( + "hour" => 1005, + "day" => 1005, + "week" => 1006, + "month" => 1007, + ) + end + end + describe "GET /api/messages" do it "returns the stored messages with canonical node references when encrypted messages are included" do import_nodes_fixture diff --git a/web/spec/federation_spec.rb b/web/spec/federation_spec.rb index 96593d5..0d3c8d9 100644 --- a/web/spec/federation_spec.rb +++ b/web/spec/federation_spec.rb @@ -24,6 +24,8 @@ require "socket" RSpec.describe PotatoMesh::App::Federation do NODES_API_PATH = "/api/nodes".freeze + STATS_API_PATH = "/api/stats".freeze + FULL_DATA_UNAVAILABLE_REASON = "full data unavailable".freeze HTTP_CONNECTION_DOUBLE = "Net::HTTPConnection".freeze subject(:federation_helpers) do @@ -294,6 +296,37 @@ RSpec.describe PotatoMesh::App::Federation do end end + def configure_remote_node_window(now) + allow(Time).to receive(:now).and_return(now) + allow(PotatoMesh::Config).to receive(:remote_instance_max_node_age).and_return(900) + end + + def stats_mapping(now:, stats_response:, full_nodes_response:, window_nodes_response: nil) + recent_cutoff = now.to_i - 900 + mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] } + attributes_list.each do |attributes| + mapping[[attributes[:domain], STATS_API_PATH]] = stats_response + mapping[[attributes[:domain], NODES_API_PATH]] = full_nodes_response + mapping[[attributes[:domain], "/api/instances"]] = [[], :instances] + next unless window_nodes_response + + mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = window_nodes_response + end + mapping + end + + def stub_ingest_fetches(mapping, capture_paths: false) + captured_paths = [] + allow(federation_helpers).to receive(:fetch_instance_json) do |host, path| + captured_paths << [host, path] if capture_paths + mapping.fetch([host, path]) { [nil, []] } + end + allow(federation_helpers).to receive(:verify_instance_signature).and_return(true) + allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil]) + allow(federation_helpers).to receive(:upsert_instance_record) + captured_paths + end + it "stops processing once the per-response limit is exceeded" do processed_domains = [] allow(federation_helpers).to receive(:upsert_instance_record) do |_db, attrs, _signature| @@ -329,102 +362,163 @@ RSpec.describe PotatoMesh::App::Federation do expect(federation_helpers.debug_messages).to include(a_string_including("crawl limit")) end - it "requests an expanded recent node window when counting remote activity" do + it "prefers /api/stats when counting remote activity" do now = Time.at(1_700_000_000) - allow(Time).to receive(:now).and_return(now) - allow(PotatoMesh::Config).to receive(:remote_instance_max_node_age).and_return(900) - recent_cutoff = now.to_i - 900 + configure_remote_node_window(now) - mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] } - attributes_list.each_with_index do |attributes, index| - mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = [node_payload, :nodes] - mapping[[attributes[:domain], NODES_API_PATH]] = [node_payload, :nodes] - mapping[[attributes[:domain], "/api/instances"]] = [[], :instances] - allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(payload_entries[index]).and_return([attributes, "signature-#{index}", nil]) - end - - captured_paths = [] - allow(federation_helpers).to receive(:fetch_instance_json) do |host, path| - captured_paths << [host, path] - mapping.fetch([host, path]) { [nil, []] } - end - allow(federation_helpers).to receive(:verify_instance_signature).and_return(true) - allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil]) - allow(federation_helpers).to receive(:upsert_instance_record) + mapping = stats_mapping( + now:, + stats_response: [{ "active_nodes" => { "hour" => 5, "day" => 7, "week" => 9, "month" => 11 }, "sampled" => false }, :stats], + full_nodes_response: [node_payload, :nodes], + ) + captured_paths = stub_ingest_fetches(mapping, capture_paths: true) federation_helpers.ingest_known_instances_from!(db, seed_domain) expect(captured_paths).to include( - [attributes_list[0][:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"], - [attributes_list[1][:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"], - [attributes_list[2][:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"], + [attributes_list[0][:domain], STATS_API_PATH], + [attributes_list[1][:domain], STATS_API_PATH], + [attributes_list[2][:domain], STATS_API_PATH], ) expect(captured_paths).to include( [attributes_list[0][:domain], NODES_API_PATH], [attributes_list[1][:domain], NODES_API_PATH], [attributes_list[2][:domain], NODES_API_PATH], ) - expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(node_payload.length)) + expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(5)) end - it "falls back to full node data when the recent window request fails" do + it "prefers recent node window counts when /api/stats is unavailable" do now = Time.at(1_700_000_000) - allow(Time).to receive(:now).and_return(now) - allow(PotatoMesh::Config).to receive(:remote_instance_max_node_age).and_return(900) - recent_cutoff = now.to_i - 900 + configure_remote_node_window(now) + full_nodes_payload = node_payload.take(2) + recent_window_payload = node_payload + recent_path = "/api/nodes?since=#{now.to_i - 900}&limit=1000" - mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] } - attributes_list.each_with_index do |attributes, index| - mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = [nil, ["no window"]] - mapping[[attributes[:domain], NODES_API_PATH]] = [node_payload, :nodes] - mapping[[attributes[:domain], "/api/instances"]] = [[], :instances] - allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(payload_entries[index]).and_return([attributes, "signature-#{index}", nil]) - end - - captured_paths = [] - allow(federation_helpers).to receive(:fetch_instance_json) do |host, path| - captured_paths << [host, path] - mapping.fetch([host, path]) { [nil, []] } - end - allow(federation_helpers).to receive(:verify_instance_signature).and_return(true) - allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil]) - allow(federation_helpers).to receive(:upsert_instance_record) + mapping = stats_mapping( + now:, + stats_response: [nil, ["stats unavailable"]], + full_nodes_response: [full_nodes_payload, :nodes], + window_nodes_response: [recent_window_payload, :nodes], + ) + captured_paths = stub_ingest_fetches(mapping, capture_paths: true) federation_helpers.ingest_known_instances_from!(db, seed_domain) + expect(captured_paths).to include( + [attributes_list[0][:domain], STATS_API_PATH], + [attributes_list[1][:domain], STATS_API_PATH], + [attributes_list[2][:domain], STATS_API_PATH], + ) expect(captured_paths).to include( [attributes_list[0][:domain], NODES_API_PATH], [attributes_list[1][:domain], NODES_API_PATH], [attributes_list[2][:domain], NODES_API_PATH], ) - expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(be_nil) + expect(captured_paths).to include( + [attributes_list[0][:domain], recent_path], + [attributes_list[1][:domain], recent_path], + [attributes_list[2][:domain], recent_path], + ) + expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(recent_window_payload.length)) end it "falls back to recent node window when full node data is unavailable" do now = Time.at(1_700_000_000) - allow(Time).to receive(:now).and_return(now) - allow(PotatoMesh::Config).to receive(:remote_instance_max_node_age).and_return(900) - recent_cutoff = now.to_i - 900 + configure_remote_node_window(now) - mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] } - attributes_list.each_with_index do |attributes, index| - mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = [node_payload, :nodes] - mapping[[attributes[:domain], NODES_API_PATH]] = [nil, ["full data unavailable"]] - mapping[[attributes[:domain], "/api/instances"]] = [[], :instances] - allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(payload_entries[index]).and_return([attributes, "signature-#{index}", nil]) - end - - allow(federation_helpers).to receive(:fetch_instance_json) do |host, path| - mapping.fetch([host, path]) { [nil, []] } - end - allow(federation_helpers).to receive(:verify_instance_signature).and_return(true) - allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil]) - allow(federation_helpers).to receive(:upsert_instance_record) + mapping = stats_mapping( + now:, + stats_response: [nil, ["stats unavailable"]], + full_nodes_response: [nil, [FULL_DATA_UNAVAILABLE_REASON]], + window_nodes_response: [node_payload, :nodes], + ) + stub_ingest_fetches(mapping) federation_helpers.ingest_known_instances_from!(db, seed_domain) expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(node_payload.length)) end + + it "uses recent node window fallback when stats succeed but full node data is unavailable" do + now = Time.at(1_700_000_000) + configure_remote_node_window(now) + recent_path = "/api/nodes?since=#{now.to_i - 900}&limit=1000" + + mapping = stats_mapping( + now:, + stats_response: [{ "active_nodes" => { "hour" => 9, "day" => 10, "week" => 11, "month" => 12 }, "sampled" => false }, :stats], + full_nodes_response: [nil, [FULL_DATA_UNAVAILABLE_REASON]], + window_nodes_response: [node_payload, :nodes], + ) + captured_paths = stub_ingest_fetches(mapping, capture_paths: true) + + federation_helpers.ingest_known_instances_from!(db, seed_domain) + + expect(captured_paths).to include( + [attributes_list[0][:domain], STATS_API_PATH], + [attributes_list[1][:domain], STATS_API_PATH], + [attributes_list[2][:domain], STATS_API_PATH], + ) + expect(captured_paths).to include( + [attributes_list[0][:domain], recent_path], + [attributes_list[1][:domain], recent_path], + [attributes_list[2][:domain], recent_path], + ) + expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(9)) + end + + it "handles URI metadata from malformed /api/stats payloads without crashing" do + now = Time.at(1_700_000_000) + configure_remote_node_window(now) + + mapping = stats_mapping( + now:, + stats_response: [{ "unexpected" => "shape" }, URI.parse("https://ally-0.mesh/api/stats")], + full_nodes_response: [node_payload.take(2), :nodes], + window_nodes_response: [node_payload, :nodes], + ) + stub_ingest_fetches(mapping) + + expect do + federation_helpers.ingest_known_instances_from!(db, seed_domain) + end.not_to raise_error + expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(node_payload.length)) + end + + it "skips remote entries when both full and window node feeds are unavailable" do + now = Time.at(1_700_000_000) + configure_remote_node_window(now) + recent_path = "/api/nodes?since=#{now.to_i - 900}&limit=1000" + + mapping = stats_mapping( + now:, + stats_response: [{ "active_nodes" => { "hour" => 3, "day" => 3, "week" => 3, "month" => 3 }, "sampled" => false }, :stats], + full_nodes_response: [nil, [FULL_DATA_UNAVAILABLE_REASON]], + window_nodes_response: [nil, ["window unavailable"]], + ) + captured_paths = stub_ingest_fetches(mapping, capture_paths: true) + upserted = [] + allow(federation_helpers).to receive(:upsert_instance_record) do |_db, attrs, _signature| + upserted << attrs + end + + federation_helpers.ingest_known_instances_from!(db, seed_domain) + + expect(captured_paths).to include( + [attributes_list[0][:domain], NODES_API_PATH], + [attributes_list[1][:domain], NODES_API_PATH], + [attributes_list[2][:domain], NODES_API_PATH], + ) + expect(captured_paths).to include( + [attributes_list[0][:domain], recent_path], + [attributes_list[1][:domain], recent_path], + [attributes_list[2][:domain], recent_path], + ) + expect(upserted).to be_empty + expect(federation_helpers.warn_messages).to include("Failed to load remote node data") + expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(3)) + end end describe ".upsert_instance_record" do