mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-08-06 17:03:27 +02:00
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
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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"])
|
||||
|
||||
@@ -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).'
|
||||
);
|
||||
});
|
||||
@@ -69,6 +69,144 @@ import {
|
||||
roleRenderOrder,
|
||||
} from './role-helpers.js';
|
||||
|
||||
/**
|
||||
* Compute active-node counts from a local node array.
|
||||
*
|
||||
* @param {Array<Object>} 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<Object>,
|
||||
* 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<typeof setTimeout>|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
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+156
-62
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user