diff --git a/CHANGELOG.md b/CHANGELOG.md index 7e3c0f57..e76a8589 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -179,6 +179,12 @@ * Web: the header and footer print `SITE_NAME` and the version as text, and asset URLs, module preloads and the import map carry the version URL-encoded (SPEC HD6) * Web: the unused `/potatomesh-logo.svg` route is removed; the static handler serves the logo as before (SPEC HD7) * Docs: `deploy/nginx.example.conf` and `DOCKER.md` clear the `Forwarded` request header, so the scheme of generated URLs comes from `X-Forwarded-Proto` (SPEC HD4) +* Web: `/api/telemetry/aggregated` and `/api/stats/activity` serve two window/bucket pairs each, the default and the one `/charts` draws (the map card uses the activity default); any other pair is served the nearest one, a bucket too small for its window is no longer a 400, and `since` is rounded down to its bucket and answered from the cached series, up to 60 s old (SPEC RC1) +* Web: telemetry, position, neighbor and trace reads filter on `rx_time` and search its index, also with `before` (SPEC RC2) +* Web: the API response cache holds collection pages only at `limit` 10, 200 or 1000, other limits are served uncached, and pages filtered by `protocol` no longer evict `/api/stats`, `/api/instances` or the unfiltered pages (SPEC RC3) +* Web: `GET /api/instances` no longer writes; duplicate instance rows are merged and the self record is refreshed at boot and on each federation announcement cycle, and this instance's own row stays listed while its newest node is under 7 days old (SPEC RC4) +* Web: API reads take `since`, `before`, `limit` and the window and bucket parameters from the query string only, never from a request body (SPEC RC6) +* Docs: `deploy/nginx.example.conf` and `DOCKER.md` carry an optional per-address rate limit for API requests with `since` or `before`, collection requests with a `limit` other than 10, 200 or 1000, and `/api/telemetry/aggregated` (SPEC RC5) * Data: the MeshCore ingestor posts each roster contact's node once per connect instead of three times, and its position once instead of twice; a re-advert of a known contact posts its node and position once each instead of twice, or twice instead of three times on firmware 1.16 and later (SPEC CU1-CU3) * Web: a federation instance row is replaced in one transaction: a failed write keeps the stored row, and two writes for one domain at once no longer fail on the unique domain index (SPEC FK1) * Web: a database error while a crawl stores one peer is logged and the crawl goes on with the next peer (SPEC FK2) diff --git a/DOCKER.md b/DOCKER.md index ff56531c..b5850255 100644 --- a/DOCKER.md +++ b/DOCKER.md @@ -176,6 +176,30 @@ Three things that bite in practice - all handled in the example file: off` and ships its own `ssl_session_cache`; duplicating the directive is a fatal config error. Leave it off. +Rate limits (optional). The example carries two commented limits: federation +peers by User-Agent, and per client address the API requests the response +cache does not answer (any `since` or `before`, or a collection `limit` other +than 10, 200 or 1000) plus `/api/telemetry/aggregated`. Both answer with the +example's one `limit_req_status 429;` line: uncomment it once. The read limit: + +```nginx +# http { } +map "$uri?$args" $potatomesh_read { + default ""; + ~^/api/telemetry/aggregated[?] $binary_remote_addr; + "~*^/api/.*[?&]((s|%73)(i|%69)(n|%6e)(c|%63)(e|%65)|(b|%62)(e|%65)(f|%66)(o|%6f)(r|%72)(e|%65))=" $binary_remote_addr; + "~*^/api/(nodes|positions|telemetry|neighbors|traces|ingestors|waypoints|messages)/?[?](.*&)?(l|%6c)(i|%69)(m|%6d)(i|%69)(t|%74)=(?!(10|200|1000)(&|$))" $binary_remote_addr; +} +limit_req_zone $potatomesh_read zone=potatomesh_reads:10m rate=40r/s; + +# location / { } +limit_req zone=potatomesh_reads burst=2400 nodelay; +``` + +An open dashboard sends up to 8 of these requests a second on a busy mesh, and +visitors behind one NAT (camps, venues) share one address: set `rate` to 8r/s +per such visitor. + Verify after `nginx -t && systemctl reload nginx`: ```bash diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index f4451fac..e61c8bcc 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -434,22 +434,47 @@ Every read endpoint enforces a server-side rolling-window floor on the data it r | --- | --- | --- | | `GET /api/nodes` | 7 days | filtered by `nodes.last_heard` | | `GET /api/messages` | 7 days | filtered by `messages.rx_time` | -| `GET /api/positions` | 7 days | filtered by `COALESCE(rx_time, position_time)` | -| `GET /api/telemetry` | 7 days | filtered by `COALESCE(rx_time, telemetry_time)` | -| `GET /api/instances` | 7 days | filtered by `instances.last_update_time` | +| `GET /api/positions` | 7 days | filtered by `positions.rx_time` | +| `GET /api/telemetry` | 7 days | filtered by `telemetry.rx_time` | +| `GET /api/instances` | 7 days | filtered by `instances.last_update_time`; this instance's own row by its newest node's `last_heard` | | `GET /api/waypoints` | 7 days | filtered by `waypoints.rx_time`; rows past their `expire` timestamp are additionally excluded from the moment of expiry (SPEC W5). 404 under `PRIVATE=1` (message-grade privacy, SPEC W3). The per-author `GET /api/waypoints/:id` (SPEC W11 - feeds the node page's Waypoints section) uses the standard per-id 28-day window and the same expiry/privacy gates. | | `GET /api/destinations` | 7 days | filtered by the owning node's `nodes.last_heard`, the `/api/nodes` window; `?node_id=` takes the per-id 28-day window. The destination's own `last_heard` is capped at 28 days on both, and the `since` clamp applies to that cap (SPEC RA8). | | `GET /api/neighbors` | 28 days | sparse data; widened to keep slow scrapes visible | | `GET /api/traces` | 28 days | sparse data; same rationale | | `GET /api/ingestors` | 28 days | sparse heartbeats; same rationale | | `GET /api/.../:id` (per-id lookup) | 28 days | every per-id route uses the extended window so callers can backfill historical context for a specific node/conversation that has dropped out of the bulk view. The `since` clamp still applies. | -| `GET /api/telemetry/aggregated` | caller-controlled | `?windowSeconds=` is mandatory; defaults to 86 400 (1 day). Bounded by `MAX_QUERY_LIMIT` on bucket count, not by a hard floor. | +| `GET /api/telemetry/aggregated` | 1 or 7 days | one of two canonical window/bucket pairs, see Bucketed series windows below. | | `GET /api/stats` | n/a | reports activity counts at fixed `hour`/`day`/`week`/`month` buckets; response shape documented below. | Federation peers should not assume an unbounded historical window: a peer that requests `/api/messages?since=0` from a partner expecting "everything" will only ever receive the last seven days. To pull older state, request the per-id endpoint (28 days) for the relevant nodes. The constants live in `web/lib/potato_mesh/config.rb` (`week_seconds`, `four_weeks_seconds`). +### Bucketed series windows (SPEC RC1) + +`GET /api/telemetry/aggregated` (`windowSeconds`, `bucketSeconds`) and `GET /api/stats/activity` (`window_seconds`, `bucket_seconds`) serve only these window/bucket pairs, in seconds: + +| Route | Pairs | +| --- | --- | +| `GET /api/telemetry/aggregated` | 86 400 / 300 (default), 604 800 / 3 600 (`/charts`) | +| `GET /api/stats/activity` | 86 400 / 3 600 (default, map card), 604 800 / 7 200 (`/charts`) | + +- Nearest pair. The requested window, clamped to 28 days, picks the pair whose window is closest to it; among pairs equally close, the bucket closest to the requested bucket decides; a remaining tie goes to the shorter window. 3 600 / 60 is served 86 400 / 300, 2 419 200 / 86 400 is served 604 800 / 3 600, 345 600 / 300 is served 86 400 / 300. +- A missing parameter takes the default pair's value. A window or bucket that is not a positive integer is a `400`; every other pair is served. +- The bucket served is each telemetry bucket's `bucket_seconds`, and each activity bucket's `bucket_end` minus `bucket_start`. +- `since` is rounded down to the start of its bucket. The answer is the cached series' buckets that start at or after it; a `since` at or before the window floor returns the whole series. +- Each pair is cached for 60 s, the telemetry pairs per private flag. +- A `since` answer comes from the cached series and can be up to 60 s old. + +### GET response cache (SPEC RC3) + +- Cached: the newest page (no `since`, no `before`) of `/api/nodes`, `/api/positions`, `/api/telemetry`, `/api/neighbors`, `/api/traces`, `/api/ingestors`, `/api/waypoints` and `/api/messages` at `limit` 10, 200 or 1000; `/api/stats`; `/api/instances`; the bucketed series pairs. +- A request to those collections at any other `limit`, or with `since` or `before`, is answered from a fresh query, as are the per-id routes and `/api/destinations`. +- `GET /api/*` takes `since`, `before`, `limit`, `windowSeconds`, `bucketSeconds`, `window_seconds` and `bucket_seconds` from the query string only; a request body does not set them (SPEC RC6). +- Pages filtered by `protocol` share 64 entries, oldest out first. They never evict an unfiltered page, `/api/stats`, `/api/instances` or a bucketed series. +- A `GET /api/instances` miss only reads. Duplicate domains are merged and the self record is refreshed at boot and on each federation announcement cycle. +- This instance's own row is listed unless its newest node was heard more than 7 days ago, with the fields of its last refresh. + ### GET endpoint backward pagination (`?before=`) The eight bulk collection endpoints - `GET /api/nodes`, `/api/positions`, @@ -698,9 +723,9 @@ One row per announced destination, newest `last_heard` first. A bucketed packets/hour series over `ingestor_activity`, feeding the mesh-activity map-card sparkline and the `/charts` activity figure. snake_case params (the -API norm): `window_seconds` (default 86 400, clamped to the 28-day floor) and -`bucket_seconds` (default 3 600); a bucket count over `MAX_QUERY_LIMIT` is a `400`. -An optional `since` bypasses the response cache. +API norm): `window_seconds` (default 86 400) and `bucket_seconds` (default 3 600), +served as the nearest canonical pair (Bucketed series windows above). An optional +`since` is rounded down to its bucket and answered from the cached series. ```jsonc [ diff --git a/deploy/nginx.example.conf b/deploy/nginx.example.conf index 989fe823..b541c9d5 100644 --- a/deploy/nginx.example.conf +++ b/deploy/nginx.example.conf @@ -38,6 +38,22 @@ types_hash_max_size 2048; # limit_req_zone $potatomesh_peer zone=potatomesh_peers:1m rate=30r/m; # limit_req_status 429; +# Optional: rate-limit reads per client address: API requests the response +# cache does not answer (any `since` or `before`, or a collection `limit` other +# than 10, 200 or 1000) and /api/telemetry/aggregated. An open dashboard sends +# up to 8 of these a second on a busy mesh, and visitors behind one NAT (camps, +# venues) share one address: set `rate` to 8r/s per such visitor. To enable, +# uncomment the map and the limit_req_zone lines here, the limit_req_status +# line above (once, for either limit or both) and the second limit_req line in +# `location /`. +# map "$uri?$args" $potatomesh_read { +# default ""; +# ~^/api/telemetry/aggregated[?] $binary_remote_addr; +# "~*^/api/.*[?&]((s|%73)(i|%69)(n|%6e)(c|%63)(e|%65)|(b|%62)(e|%65)(f|%66)(o|%6f)(r|%72)(e|%65))=" $binary_remote_addr; +# "~*^/api/(nodes|positions|telemetry|neighbors|traces|ingestors|waypoints|messages)/?[?](.*&)?(l|%6c)(i|%69)(m|%6d)(i|%69)(t|%74)=(?!(10|200|1000)(&|$))" $binary_remote_addr; +# } +# limit_req_zone $potatomesh_read zone=potatomesh_reads:10m rate=40r/s; + # --------------------------------------------------------------------------- # server { } context # --------------------------------------------------------------------------- @@ -96,6 +112,8 @@ server { location / { # Optional peer rate limit, see the map near the top of this file. # limit_req zone=potatomesh_peers burst=20 nodelay; + # Optional read rate limit, see the second map near the top of this file. + # limit_req zone=potatomesh_reads burst=2400 nodelay; proxy_pass http://potatomesh; proxy_http_version 1.1; # required for upstream keepalive diff --git a/web/lib/potato_mesh/application.rb b/web/lib/potato_mesh/application.rb index fb2e7b58..f9d76208 100644 --- a/web/lib/potato_mesh/application.rb +++ b/web/lib/potato_mesh/application.rb @@ -59,6 +59,7 @@ require_relative "application/meshtastic/payload_decoder" require_relative "application/data_processing" require_relative "application/filesystem" require_relative "application/api_cache" +require_relative "application/aggregate_windows" require_relative "application/asset_cache_control" require_relative "application/security_headers" require_relative "application/public_scheme" @@ -262,6 +263,9 @@ module PotatoMesh log_instance_domain_resolution log_instance_public_key refresh_well_known_document_if_stale + # The instance housekeeping runs here and on each federation + # announcement cycle, so GET /api/instances only reads (SPEC RC4). + clean_duplicate_instances! ensure_self_instance_record! update_all_prometheus_metrics_from_nodes diff --git a/web/lib/potato_mesh/application/aggregate_windows.rb b/web/lib/potato_mesh/application/aggregate_windows.rb new file mode 100644 index 00000000..cfc75679 --- /dev/null +++ b/web/lib/potato_mesh/application/aggregate_windows.rb @@ -0,0 +1,89 @@ +# 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. + +# frozen_string_literal: true + +module PotatoMesh + module App + # Canonical window/bucket pairs of the two bucketed read endpoints, + # +GET /api/telemetry/aggregated+ and +GET /api/stats/activity+ + # (SPEC RC1). + # + # Each endpoint serves only its pairs below: the route default and the + # pairs the frontend charts request. A request for any other pair is + # served the nearest pair ({.nearest_pair}), so the response cache holds + # at most two series per endpoint (and private flag) however many + # distinct windows clients send. A +since+ is rounded down to the start + # of its bucket ({.bucket_floor}) and answered from the same cached + # series ({.series_since}): a bucket that starts at or after it holds the + # same rows in the full series as in a query bounded by it. + module AggregateWindows + # Pairs of +/api/telemetry/aggregated+ as +[window, bucket]+ seconds: + # the route default, one day in 5-minute buckets, and the /charts + # telemetry figures, seven days in 1-hour buckets (+charts-page.js+). + TELEMETRY_PAIRS = [ + [Queries::DEFAULT_TELEMETRY_WINDOW_SECONDS, Queries::DEFAULT_TELEMETRY_BUCKET_SECONDS].freeze, + [604_800, 3_600].freeze, + ].freeze + + # Pairs of +/api/stats/activity+: the route default and map-card + # sparkline, one day in 1-hour buckets (+stats.js+), and the /charts + # activity figure, seven days in 2-hour buckets + # (+mesh-activity-chart.js+). + ACTIVITY_PAIRS = [ + [Queries::DEFAULT_ACTIVITY_WINDOW_SECONDS, Queries::DEFAULT_ACTIVITY_BUCKET_SECONDS].freeze, + [604_800, 7_200].freeze, + ].freeze + + module_function + + # The canonical pair nearest a requested window and bucket: the pair + # whose window is closest to +window_seconds+, then, among pairs equally + # close, the one whose bucket is closest to +bucket_seconds+; a pair + # still tied goes to the one listed first, the shorter window. + # + # @param pairs [Array] canonical + # +[window, bucket]+ pairs, shortest window first. + # @param window_seconds [Integer] requested window, already clamped to + # the 28-day visibility floor. + # @param bucket_seconds [Integer] requested bucket width. + # @return [Array(Integer, Integer)] the canonical +[window, bucket]+. + def nearest_pair(pairs, window_seconds, bucket_seconds) + pairs.each_with_index.min_by do |(window, bucket), index| + [(window - window_seconds).abs, (bucket - bucket_seconds).abs, index] + end.first + end + + # Round a timestamp down to the start of its bucket, the boundary the + # bucketed queries group on (+(ts / bucket) * bucket+). + # + # @param timestamp [Integer] unix seconds. + # @param bucket_seconds [Integer] bucket width. + # @return [Integer] the start of the bucket holding +timestamp+. + def bucket_floor(timestamp, bucket_seconds) + (timestamp / bucket_seconds) * bucket_seconds + end + + # Reduce a cached series to the buckets that start at or after +since+. + # + # @param json [String] serialised series, an array of buckets that each + # carry +bucket_start+. + # @param since [Integer] first bucket start to keep, a bucket boundary. + # @return [String] the reduced series as JSON. + def series_since(json, since) + JSON.generate(JSON.parse(json).select { |bucket| bucket["bucket_start"].to_i >= since }) + end + end + end +end diff --git a/web/lib/potato_mesh/application/api_cache.rb b/web/lib/potato_mesh/application/api_cache.rb index 37eee668..a42be59b 100644 --- a/web/lib/potato_mesh/application/api_cache.rb +++ b/web/lib/potato_mesh/application/api_cache.rb @@ -23,20 +23,33 @@ module PotatoMesh # Each entry is stored with a monotonic expiration time and a pre-computed # ETag so the route handler can skip recomputing the digest on cache hits. # - # The cache is bounded to {MAX_ENTRIES} to prevent unbounded memory growth - # from attacker-controlled query parameters. When the limit is reached the - # oldest entry by insertion order is evicted (LRU-ish via Ruby hash ordering). + # Entries come in two tiers (SPEC RC3). A *fixed* entry is one of a + # finite set the routes name: +/api/stats+, +/api/instances+, the + # canonical bucketed series and each collection's page without a + # +protocol+ filter at a limit of {CACHEABLE_LIMITS}. Every other cached + # page, one filtered by +protocol+, is a *variant*. Each tier holds at + # most {MAX_ENTRIES} entries, evicting its own oldest entry by insertion + # order, so no number of variants can evict a fixed entry. A route serves + # a request whose limit is outside {CACHEABLE_LIMITS} uncached, so free + # query parameters never reach the keys. # # Invalidation can target a specific prefix (e.g. +"api:nodes:"+) so that an # ingest POST to +/api/messages+ does not flush the neighbors cache. # A single-flight guard coalesces concurrent misses for the same key so only # one thread computes the value while others wait for the result. module ApiCache - # Hard cap on the number of cached entries to prevent memory exhaustion. - # With the whitelisted protocol values and known limit set, the realistic - # key space is ~30 entries. 64 provides generous headroom. + # Cap on the entries of each tier, to bound memory. The fixed tier + # names at most 55 keys, both private-flag values counted, so it never + # evicts in practice; variants evict each other. MAX_ENTRIES = 64 + # The collection limits the app itself requests, the only ones whose + # pages are cached: 10, the federation node list peers fetch + # (+remote_instance_min_node_count+, SPEC FL2) and the Matrix bridge's + # fallback; 200, the default when +limit+ is absent and the trace page; + # 1000, +MAX_QUERY_LIMIT+, the dashboard's pages. + CACHEABLE_LIMITS = [10, 200, 1000].freeze + @store = {} @inflight = {} @mutex = Mutex.new @@ -54,10 +67,12 @@ module PotatoMesh # @param key [String] cache key incorporating all relevant query # parameters (limit, protocol, etc.). # @param ttl_seconds [Numeric] time-to-live for the cached entry. + # @param fixed [Boolean] whether +key+ belongs to the finite set of + # fixed entries that no variant evicts (SPEC RC3). # @yield Computes the value to cache when the entry is missing or # expired. The block should return the serialised JSON string. # @return [Hash{Symbol => String}] +:value+ and +:etag+ of the response. - def fetch(key, ttl_seconds:) + def fetch(key, ttl_seconds:, fixed: false) now = monotonic_now @mutex.synchronize do @@ -86,8 +101,11 @@ module PotatoMesh etag = Digest::MD5.hexdigest(value) @mutex.synchronize do - evict_oldest_if_full - @store[key] = { value: value, etag: etag, expires_at: monotonic_now + ttl_seconds } + # A recomputed key replaces its expired entry, which leaves its + # tier first, so it evicts no other entry. + @store.delete(key) + evict_oldest_if_full(fixed) + @store[key] = { value: value, etag: etag, expires_at: monotonic_now + ttl_seconds, fixed: fixed } cv = @inflight.delete(key) cv&.broadcast end @@ -141,6 +159,15 @@ module PotatoMesh @mutex.synchronize { @store.size } end + # Whether a collection page at +limit+ may be cached: only the limits + # the app requests are, so a client cannot mint keys (SPEC RC3). + # + # @param limit [Integer] the coerced request limit. + # @return [Boolean] true when +limit+ is one of {CACHEABLE_LIMITS}. + def cacheable_limit?(limit) + CACHEABLE_LIMITS.include?(limit) + end + private # Use the monotonic clock so TTL calculations are immune to wall-clock @@ -149,13 +176,16 @@ module PotatoMesh Process.clock_gettime(Process::CLOCK_MONOTONIC) end - # Evict the oldest entry when the store is at capacity. Ruby hashes - # preserve insertion order, so +first+ is the oldest key. - def evict_oldest_if_full - while @store.size >= MAX_ENTRIES - oldest_key = @store.each_key.first - @store.delete(oldest_key) - end + # Evict the oldest entries of one tier until it has room for one more. + # Ruby hashes preserve insertion order, so the first keys are the + # oldest; the other tier is never touched (SPEC RC3). + # + # @param fixed [Boolean] the tier of the entry about to be stored. + # @return [void] + def evict_oldest_if_full(fixed) + tier = @store.each_key.select { |key| @store[key][:fixed] == fixed } + excess = tier.size - MAX_ENTRIES + 1 + tier.first(excess).each { |key| @store.delete(key) } if excess.positive? end end end diff --git a/web/lib/potato_mesh/application/federation/announce.rb b/web/lib/potato_mesh/application/federation/announce.rb index fe270db4..a0a98331 100644 --- a/web/lib/potato_mesh/application/federation/announce.rb +++ b/web/lib/potato_mesh/application/federation/announce.rb @@ -149,6 +149,9 @@ module PotatoMesh return unless federation_enabled? return if federation_shutdown_requested? + # The instance housekeeping GET /api/instances no longer does: merge + # duplicate domains, then refresh the self record (SPEC RC4). + clean_duplicate_instances! attributes, signature = ensure_self_instance_record! payload_json = JSON.generate(instance_announcement_payload(attributes, signature)) domains = federation_target_domains(attributes[:domain]) diff --git a/web/lib/potato_mesh/application/instances.rb b/web/lib/potato_mesh/application/instances.rb index b0e1750c..89fec11a 100644 --- a/web/lib/potato_mesh/application/instances.rb +++ b/web/lib/potato_mesh/application/instances.rb @@ -171,15 +171,20 @@ module PotatoMesh # Fetch all instance rows ready to be served by the API while handling # malformed rows gracefully. The dataset is restricted to records updated # within the rolling window defined by PotatoMesh::Config.week_seconds. + # Reads only: {#clean_duplicate_instances!} runs at boot and on each + # announcement cycle instead (SPEC RC4). # # @return [Array] list of cleaned instance payloads. def load_instances_for_api - clean_duplicate_instances! - - db = open_database(readonly: true) - db.results_as_hash = true now = Time.now.to_i min_last_update_time = now - PotatoMesh::Config.week_seconds + self_id = app_constant(:SELF_INSTANCE_ID) + # Our own row is stored at boot and on each announcement cycle only, so + # its time can trail a node heard since: it is listed by its newest + # node, read now, as when every GET refreshed it first (SPEC RC4). + self_listed = (latest_node_update_timestamp || now) >= min_last_update_time + db = open_database(readonly: true) + db.results_as_hash = true sql = <<~SQL SELECT id, domain, pubkey, name, version, channel, frequency, latitude, longitude, last_update_time, is_private, nodes_count, @@ -188,20 +193,24 @@ module PotatoMesh FROM instances WHERE domain IS NOT NULL AND TRIM(domain) != '' AND pubkey IS NOT NULL AND TRIM(pubkey) != '' - AND last_update_time IS NOT NULL AND last_update_time >= ? + AND (id = ? OR (last_update_time IS NOT NULL AND last_update_time >= ?)) ORDER BY LOWER(domain) SQL rows = with_busy_retry do - db.execute(sql, min_last_update_time) + db.execute(sql, [self_id, min_last_update_time]) end rows.each_with_object([]) do |row, memo| normalized = normalize_instance_row(row) next unless normalized - last_update_time = normalized["last_update"] - next unless last_update_time.is_a?(Integer) && last_update_time >= min_last_update_time + if normalized["id"] == self_id + next unless self_listed + else + last_update_time = normalized["last_update"] + next unless last_update_time.is_a?(Integer) && last_update_time >= min_last_update_time + end memo << normalized end diff --git a/web/lib/potato_mesh/application/queries/federation_queries.rb b/web/lib/potato_mesh/application/queries/federation_queries.rb index b56bbf77..7e958cd3 100644 --- a/web/lib/potato_mesh/application/queries/federation_queries.rb +++ b/web/lib/potato_mesh/application/queries/federation_queries.rb @@ -38,7 +38,9 @@ module PotatoMesh # to twenty-eight days for backfill of historical track data. since_floor = node_ref ? now - PotatoMesh::Config.four_weeks_seconds : now - PotatoMesh::Config.week_seconds since_threshold = normalize_since_threshold(since, floor: since_floor) - where_clauses << "COALESCE(rx_time, position_time, 0) >= ?" + # +rx_time+ is NOT NULL (data/positions.sql): the window searches + # idx_positions_rx_time, with both ends under +before+ (SPEC RC2). + where_clauses << "rx_time >= ?" params << since_threshold if node_ref @@ -110,7 +112,9 @@ module PotatoMesh # for both bulk and per-id queries. min_rx_time = now - PotatoMesh::Config.four_weeks_seconds since_threshold = normalize_since_threshold(since, floor: min_rx_time) - where_clauses << "COALESCE(rx_time, 0) >= ?" + # +rx_time+ is NOT NULL (data/neighbors.sql): the window searches + # idx_neighbors_rx_time, with both ends under +before+ (SPEC RC2). + where_clauses << "rx_time >= ?" params << since_threshold if node_ref @@ -179,7 +183,9 @@ module PotatoMesh now = Time.now.to_i min_rx_time = now - PotatoMesh::Config.four_weeks_seconds since_threshold = normalize_since_threshold(since, floor: min_rx_time) - where_clauses << "COALESCE(rx_time, 0) >= ?" + # +rx_time+ is NOT NULL (data/traces.sql): the window searches + # idx_traces_rx_time, with both ends under +before+ (SPEC RC2). + where_clauses << "rx_time >= ?" params << since_threshold # The hops of an opted-out node, and in private mode of a diff --git a/web/lib/potato_mesh/application/queries/ingestor_queries.rb b/web/lib/potato_mesh/application/queries/ingestor_queries.rb index 16204915..2778990c 100644 --- a/web/lib/potato_mesh/application/queries/ingestor_queries.rb +++ b/web/lib/potato_mesh/application/queries/ingestor_queries.rb @@ -111,9 +111,9 @@ module PotatoMesh # bucket's hour-span to a rate. Every known protocol # (+meshcore+/+meshtastic+/+reticulum+) is emitted as a key, mirroring # {#query_packets_per_hour}. The window is clamped to the - # 28-day visibility floor (C4); the route pre-validates the bucket count - # against +MAX_QUERY_LIMIT+ and this query caps the row count too. Each - # stored row counts at most + # 28-day visibility floor (C4); the route serves canonical pairs only, + # at most 85 buckets (SPEC RC1), and this query caps the row count at + # +MAX_QUERY_LIMIT+ too. Each stored row counts at most # {DataProcessing::FieldLimits::MAX_HEARTBEAT_PACKETS}, as in # {#query_packets_per_hour} (SPEC IB4). # diff --git a/web/lib/potato_mesh/application/queries/telemetry_queries.rb b/web/lib/potato_mesh/application/queries/telemetry_queries.rb index 15ce8163..eb644d76 100644 --- a/web/lib/potato_mesh/application/queries/telemetry_queries.rb +++ b/web/lib/potato_mesh/application/queries/telemetry_queries.rb @@ -38,7 +38,10 @@ module PotatoMesh # to twenty-eight days so historical chart data remains reachable. since_floor = node_ref ? now - PotatoMesh::Config.four_weeks_seconds : now - PotatoMesh::Config.week_seconds since_threshold = normalize_since_threshold(since, floor: since_floor) - where_clauses << "COALESCE(rx_time, telemetry_time, 0) >= ?" + # +rx_time+ is NOT NULL (data/telemetry.sql), so the window bounds the + # column itself and the query searches idx_telemetry_rx_time; with + # +before+ that search has both ends (SPEC RC2). + where_clauses << "rx_time >= ?" params << since_threshold if node_ref @@ -163,12 +166,14 @@ module PotatoMesh node_filters = [opt_out_node_id_filter("node_id")] node_filters << hidden_client_node_id_filter("node_id") if private_mode? + # +rx_time+ is NOT NULL, so the window bounds the column itself and + # the aggregate searches idx_telemetry_rx_time instead of scanning + # the table (SPEC RC2). sql = <<~SQL SELECT #{select_clauses.join(",\n ")} FROM telemetry - WHERE COALESCE(rx_time, telemetry_time) IS NOT NULL - AND COALESCE(rx_time, telemetry_time, 0) >= ? + WHERE rx_time >= ? AND #{node_filters.join(" AND ")} GROUP BY bucket_start ORDER BY bucket_start ASC diff --git a/web/lib/potato_mesh/application/routes/api.rb b/web/lib/potato_mesh/application/routes/api.rb index fd872fef..215f3710 100644 --- a/web/lib/potato_mesh/application/routes/api.rb +++ b/web/lib/potato_mesh/application/routes/api.rb @@ -22,6 +22,11 @@ module PotatoMesh # prevent attacker-controlled strings from polluting the cache keyspace. KNOWN_PROTOCOLS = Set.new(%w[meshcore meshtastic reticulum]).freeze + # Read parameters that change a request's cost: a GET takes them from + # the query string only, never from a form body, so a proxy limiting + # on the query string sees every one (SPEC RC6). + QUERY_ONLY_PARAMS = %w[since before limit windowSeconds bucketSeconds window_seconds bucket_seconds].freeze + # Register read-only API endpoints that expose cached mesh data and # instance metadata. Invoked by Sinatra during extension registration. # @@ -29,6 +34,7 @@ module PotatoMesh # @return [void] def self.registered(app) known_protocols = KNOWN_PROTOCOLS + query_only_params = QUERY_ONLY_PARAMS app.helpers do # Sanitise the protocol query parameter to a known value. @@ -44,6 +50,56 @@ module PotatoMesh visibility = private_mode? ? :private : :public cache_control visibility, :must_revalidate, max_age: max_age end + + # Whether a collection page at +limit+ is served uncached: only the + # limits the app requests are cached, so a client cannot mint cache + # keys with free +limit+ values (SPEC RC3). + define_method(:uncached_limit?) do |limit| + !PotatoMesh::App::ApiCache.cacheable_limit?(limit) + end + + # Serve one canonical bucketed series through the response cache + # (SPEC RC1). The key names a canonical pair, a fixed entry. A + # +since+ is rounded down to the start of its bucket and answered + # from that entry with the buckets that start at or after it; a + # +since+ at or before the window floor gets the whole series, as a + # request without one does. + # + # @param cache_key [String] the canonical pair's cache key. + # @param window_seconds [Integer] the canonical window. + # @param bucket_seconds [Integer] the canonical bucket width. + # @yield computes the series JSON on a cache miss. + # @return [String] the response body. + define_method(:serve_bucketed_series) do |cache_key, window_seconds:, bucket_seconds:, &compute| + cached = PotatoMesh::App::ApiCache.fetch(cache_key, ttl_seconds: 60, fixed: true, &compute) + since = coerce_integer(params["since"]) || 0 + first_bucket = PotatoMesh::App::AggregateWindows.bucket_floor(since, bucket_seconds) + if since.positive? && first_bucket > Time.now.to_i - window_seconds + json_body = PotatoMesh::App::AggregateWindows.series_since(cached[:value], first_bucket) + etag Digest::MD5.hexdigest(json_body), kind: :weak + else + json_body = cached[:value] + etag cached[:etag], kind: :weak + end + api_cache_control(max_age: 30) + json_body + end + end + + # Rack merges a form body into +params+, the body winning on a + # clash; a read takes its cost parameters from the query string + # alone, which the nginx read limit sees (SPEC RC6). + app.before "/api/*" do + next unless request.get? || request.head? + + query = request.GET + query_only_params.each do |name| + if query.key?(name) + params[name] = query[name] + else + params.delete(name) + end + end end app.before "/api/messages*" do @@ -105,13 +161,14 @@ module PotatoMesh before = coerce_positive_or_nil(params["before"]) priv = private_mode? ? 1 : 0 - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_nodes(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:nodes:#{limit}:#{protocol}:#{priv}", ttl_seconds: 15) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:nodes:#{limit}:#{protocol}:#{priv}", ttl_seconds: 15, fixed: protocol.nil?) do query_nodes(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -124,7 +181,7 @@ module PotatoMesh content_type :json priv = private_mode? ? 1 : 0 cached = PotatoMesh::App::ApiCache.fetch( - "api:stats:#{priv}", ttl_seconds: PotatoMesh::Config.stats_cache_ttl_seconds, + "api:stats:#{priv}", ttl_seconds: PotatoMesh::Config.stats_cache_ttl_seconds, fixed: true, ) do # Scope → metric → window tree (SPEC S1). The MA4 packets/hour # moving average is folded in as an additive +packets+ metric under @@ -176,27 +233,16 @@ module PotatoMesh # reach past the retention cap (C4); the query repeats the clamp. window_seconds = clamp_window_seconds(window_seconds) - bucket_count = (window_seconds.to_f / bucket_seconds).ceil - if bucket_count > PotatoMesh::App::Queries::MAX_QUERY_LIMIT - halt 400, { error: "bucket_seconds too small for requested window" }.to_json - end - - since = params["since"] - since_val = coerce_integer(since) || 0 - - if since_val > 0 - json_body = query_activity_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds, since: since).to_json - etag Digest::MD5.hexdigest(json_body), kind: :weak - api_cache_control(max_age: 30) - json_body - else - cache_key = "api:stats_activity:#{window_seconds}:#{bucket_seconds}" - cached = PotatoMesh::App::ApiCache.fetch(cache_key, ttl_seconds: 60) do - query_activity_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds, since: since).to_json - end - etag cached[:etag], kind: :weak - api_cache_control(max_age: 30) - cached[:value] + # Serve the nearest canonical pair (SPEC RC1): one cache entry per + # pair whatever the caller sends, at most 85 buckets. + window_seconds, bucket_seconds = PotatoMesh::App::AggregateWindows.nearest_pair( + PotatoMesh::App::AggregateWindows::ACTIVITY_PAIRS, window_seconds, bucket_seconds, + ) + serve_bucketed_series( + "api:stats_activity:#{window_seconds}:#{bucket_seconds}", + window_seconds: window_seconds, bucket_seconds: bucket_seconds, + ) do + query_activity_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds).to_json end end @@ -231,13 +277,14 @@ module PotatoMesh # Private mode drops CLIENT_HIDDEN rows (SPEC HC1), so the flag keys the cache. priv = private_mode? ? 1 : 0 - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_ingestors(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:ingestors:#{limit}:#{protocol}:#{priv}", ttl_seconds: 30) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:ingestors:#{limit}:#{protocol}:#{priv}", ttl_seconds: 30, fixed: protocol.nil?) do query_ingestors(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -278,13 +325,14 @@ module PotatoMesh protocol = sanitize_protocol(params["protocol"]) enc_key = include_encrypted ? "1" : "0" - if since > 0 || before + if since > 0 || before || uncached_limit?(limit) json_body = query_messages(limit, include_encrypted: include_encrypted, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:messages:#{limit}:#{enc_key}:#{protocol}", ttl_seconds: 10) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:messages:#{limit}:#{enc_key}:#{protocol}", ttl_seconds: 10, fixed: protocol.nil?) do query_messages(limit, include_encrypted: include_encrypted, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -324,13 +372,14 @@ module PotatoMesh # Private mode drops CLIENT_HIDDEN rows (SPEC HC1), so the flag keys the cache. priv = private_mode? ? 1 : 0 - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_positions(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:positions:#{limit}:#{protocol}:#{priv}", ttl_seconds: 15) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:positions:#{limit}:#{protocol}:#{priv}", ttl_seconds: 15, fixed: protocol.nil?) do query_positions(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -359,13 +408,14 @@ module PotatoMesh # Backward-pagination cursor (SPEC BP1); bypasses the cache like +since+. before = coerce_positive_or_nil(params["before"]) - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_waypoints(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:waypoints:#{limit}:#{protocol}", ttl_seconds: 15) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:waypoints:#{limit}:#{protocol}", ttl_seconds: 15, fixed: protocol.nil?) do query_waypoints(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -404,13 +454,14 @@ module PotatoMesh # Private mode drops CLIENT_HIDDEN rows (SPEC HC1), so the flag keys the cache. priv = private_mode? ? 1 : 0 - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_neighbors(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:neighbors:#{limit}:#{protocol}:#{priv}", ttl_seconds: 30) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:neighbors:#{limit}:#{protocol}:#{priv}", ttl_seconds: 30, fixed: protocol.nil?) do query_neighbors(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -441,13 +492,14 @@ module PotatoMesh # Private mode drops CLIENT_HIDDEN rows (SPEC HC1), so the flag keys the cache. priv = private_mode? ? 1 : 0 - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_telemetry(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:telemetry:#{limit}:#{protocol}:#{priv}", ttl_seconds: 15) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:telemetry:#{limit}:#{protocol}:#{priv}", ttl_seconds: 15, fixed: protocol.nil?) do query_telemetry(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -485,29 +537,19 @@ module PotatoMesh # clamp for defence in depth. window_seconds = clamp_window_seconds(window_seconds) - bucket_count = (window_seconds.to_f / bucket_seconds).ceil - if bucket_count > PotatoMesh::App::Queries::MAX_QUERY_LIMIT - halt 400, { error: "bucketSeconds too small for requested window" }.to_json - end - - since = params["since"] - since_val = coerce_integer(since) || 0 + # Serve the nearest canonical pair (SPEC RC1): one cache entry per + # pair and private flag whatever the caller sends, at most 289 + # buckets, each stating the +bucket_seconds+ served. + window_seconds, bucket_seconds = PotatoMesh::App::AggregateWindows.nearest_pair( + PotatoMesh::App::AggregateWindows::TELEMETRY_PAIRS, window_seconds, bucket_seconds, + ) # Private mode drops CLIENT_HIDDEN rows (SPEC HC1), so the flag keys the cache. priv = private_mode? ? 1 : 0 - - if since_val > 0 - json_body = query_telemetry_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds, since: since).to_json - etag Digest::MD5.hexdigest(json_body), kind: :weak - api_cache_control(max_age: 30) - json_body - else - cache_key = "api:telemetry_agg:#{window_seconds}:#{bucket_seconds}:#{priv}" - cached = PotatoMesh::App::ApiCache.fetch(cache_key, ttl_seconds: 60) do - query_telemetry_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds, since: since).to_json - end - etag cached[:etag], kind: :weak - api_cache_control(max_age: 30) - cached[:value] + serve_bucketed_series( + "api:telemetry_agg:#{window_seconds}:#{bucket_seconds}:#{priv}", + window_seconds: window_seconds, bucket_seconds: bucket_seconds, + ) do + query_telemetry_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds).to_json end end @@ -533,13 +575,14 @@ module PotatoMesh # Private mode drops CLIENT_HIDDEN rows (SPEC HC1), so the flag keys the cache. priv = private_mode? ? 1 : 0 - if since_val > 0 || before + if since_val > 0 || before || uncached_limit?(limit) json_body = query_traces(limit, since: since, before: before, protocol: protocol).to_json etag Digest::MD5.hexdigest(json_body), kind: :weak api_cache_control json_body else - cached = PotatoMesh::App::ApiCache.fetch("api:traces:#{limit}:#{protocol}:#{priv}", ttl_seconds: 30) do + # A page without a protocol filter is a fixed entry (SPEC RC3). + cached = PotatoMesh::App::ApiCache.fetch("api:traces:#{limit}:#{protocol}:#{priv}", ttl_seconds: 30, fixed: protocol.nil?) do query_traces(limit, since: since, protocol: protocol).to_json end etag cached[:etag], kind: :weak @@ -566,14 +609,13 @@ module PotatoMesh content_type :json # The federation banner is rendered on every page navigation, which # caused this endpoint to fire ~7 times in a few seconds while the - # user clicked through the site. Cache the response (including the - # self-record refresh) for a short window so navigation feels free - # without delaying signature/peer updates by more than a few - # seconds. The dedicated announcer thread keeps the underlying - # record fresh on its own cadence regardless of cache hits. + # user clicked through the site. Cache the response for a short + # window so navigation feels free without delaying peer updates by + # more than a few seconds. A miss only reads: the boot and each + # announcement cycle deduplicate the rows and refresh the self + # record (SPEC RC4). priv = private_mode? ? 1 : 0 - cached = PotatoMesh::App::ApiCache.fetch("api:instances:#{priv}", ttl_seconds: 30) do - ensure_self_instance_record! + cached = PotatoMesh::App::ApiCache.fetch("api:instances:#{priv}", ttl_seconds: 30, fixed: true) do JSON.generate(load_instances_for_api) end etag cached[:etag], kind: :weak diff --git a/web/spec/aggregate_windows_spec.rb b/web/spec/aggregate_windows_spec.rb new file mode 100644 index 00000000..a6e9f502 --- /dev/null +++ b/web/spec/aggregate_windows_spec.rb @@ -0,0 +1,88 @@ +# 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. + +# frozen_string_literal: true + +require "spec_helper" +require "json" + +RSpec.describe "PotatoMesh::App::AggregateWindows (SPEC RC1)" do + let(:windows) { PotatoMesh::App::AggregateWindows } + + describe "canonical pairs" do + it "lists the defaults and the pairs the frontend charts request" do + expect(windows::TELEMETRY_PAIRS).to eq([[86_400, 300], [604_800, 3_600]]) + expect(windows::ACTIVITY_PAIRS).to eq([[86_400, 3_600], [604_800, 7_200]]) + end + end + + describe ".nearest_pair" do + { + [3_600, 60] => [86_400, 300], + [86_400, 300] => [86_400, 300], + [86_400, 3_600] => [86_400, 300], + [345_600, 300] => [86_400, 300], + [345_601, 300] => [604_800, 3_600], + [604_800, 300] => [604_800, 3_600], + [2_419_200, 86_400] => [604_800, 3_600], + # Equally near in window and in bucket: the pair listed first. + [345_600, 1_950] => [86_400, 300], + }.each do |(window, bucket), expected| + it "maps #{window} s in #{bucket} s buckets to #{expected.join(" s in ")} s buckets for telemetry" do + expect(windows.nearest_pair(windows::TELEMETRY_PAIRS, window, bucket)).to eq(expected) + end + end + + { + [3_600, 600] => [86_400, 3_600], + [604_800, 3_600] => [604_800, 7_200], + [2_419_200, 86_400] => [604_800, 7_200], + [345_600, 5_400] => [86_400, 3_600], + }.each do |(window, bucket), expected| + it "maps #{window} s in #{bucket} s buckets to #{expected.join(" s in ")} s buckets for activity" do + expect(windows.nearest_pair(windows::ACTIVITY_PAIRS, window, bucket)).to eq(expected) + end + end + + it "decides on the bucket between pairs of one window" do + pairs = [[86_400, 300], [86_400, 3_600]] + expect(windows.nearest_pair(pairs, 86_400, 3_000)).to eq([86_400, 3_600]) + expect(windows.nearest_pair(pairs, 86_400, 1_000)).to eq([86_400, 300]) + end + end + + describe ".bucket_floor" do + it "rounds a timestamp down to the start of its bucket" do + expect(windows.bucket_floor(1_791_000_123, 300)).to eq(1_791_000_000) + expect(windows.bucket_floor(1_791_000_000, 300)).to eq(1_791_000_000) + expect(windows.bucket_floor(1_791_007_199, 7_200)).to eq(1_791_000_000) + end + end + + describe ".series_since" do + let(:series) do + [ + { "bucket_start" => 100, "bucket_end" => 200, "total" => 1 }, + { "bucket_start" => 200, "bucket_end" => 300, "total" => 2 }, + { "bucket_start" => 300, "bucket_end" => 400, "total" => 3 }, + ].to_json + end + + it "keeps the buckets that start at or after since" do + expect(JSON.parse(windows.series_since(series, 200)).map { |b| b["total"] }).to eq([2, 3]) + expect(JSON.parse(windows.series_since(series, 100))).to eq(JSON.parse(series)) + expect(windows.series_since(series, 400)).to eq("[]") + end + end +end diff --git a/web/spec/api_cache_spec.rb b/web/spec/api_cache_spec.rb index e664776a..c9a43cd5 100644 --- a/web/spec/api_cache_spec.rb +++ b/web/spec/api_cache_spec.rb @@ -134,6 +134,47 @@ RSpec.describe PotatoMesh::App::ApiCache do end end + describe "fixed entries (SPEC RC3)" do + let(:max) { described_class::MAX_ENTRIES } + + it "never evicts a fixed entry to make room for a variant" do + described_class.fetch("api:stats:0", ttl_seconds: 60, fixed: true) { "stats" } + (max + 8).times { |i| described_class.fetch("variant:#{i}", ttl_seconds: 60) { "v#{i}" } } + + expect(described_class.fetch("api:stats:0", ttl_seconds: 60, fixed: true) { "recomputed" }[:value]).to eq("stats") + expect(described_class.size).to eq(max + 1) + # The variants evict each other, oldest first. + expect(described_class.fetch("variant:7", ttl_seconds: 60) { "recomputed" }[:value]).to eq("recomputed") + expect(described_class.fetch("variant:#{max + 7}", ttl_seconds: 60) { "recomputed" }[:value]).to eq("v#{max + 7}") + end + + it "holds at most MAX_ENTRIES fixed entries, the oldest out first" do + (max + 1).times { |i| described_class.fetch("fixed:#{i}", ttl_seconds: 60, fixed: true) { "f#{i}" } } + + expect(described_class.size).to eq(max) + expect(described_class.fetch("fixed:#{max}", ttl_seconds: 60, fixed: true) { "recomputed" }[:value]).to eq("f#{max}") + expect(described_class.fetch("fixed:0", ttl_seconds: 60, fixed: true) { "recomputed" }[:value]).to eq("recomputed") + end + + it "evicts nothing else when an expired entry is recomputed" do + described_class.fetch("variant:first", ttl_seconds: 60) { "first" } + described_class.fetch("variant:expiring", ttl_seconds: 0) { "old" } + (max - 2).times { |i| described_class.fetch("variant:#{i}", ttl_seconds: 60) { "v#{i}" } } + sleep 0.01 + + expect(described_class.fetch("variant:expiring", ttl_seconds: 60) { "new" }[:value]).to eq("new") + expect(described_class.size).to eq(max) + expect(described_class.fetch("variant:first", ttl_seconds: 60) { "recomputed" }[:value]).to eq("first") + end + end + + describe ".cacheable_limit?" do + it "accepts only the limits the app requests" do + expect([10, 200, 1000].map { |limit| described_class.cacheable_limit?(limit) }).to eq([true, true, true]) + expect([1, 9, 11, 199, 201, 999].map { |limit| described_class.cacheable_limit?(limit) }).to all(be(false)) + end + end + describe ".size" do it "reports the number of cached entries" do expect(described_class.size).to eq(0) diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 79eacd09..5fef8760 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -3129,7 +3129,7 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(payload).not_to be_empty end - it "deduplicates records by domain keeping the newest entry" do + it "deduplicates records by domain on the announcement cycle, keeping the newest entry (SPEC RC4)" do newer_time = Time.now.to_i older_time = newer_time - 60 @@ -3177,6 +3177,15 @@ RSpec.describe "Potato Mesh Sinatra app" do ) end + # A GET only reads: both rows are served until the cycle merges them. + get "/api/instances" + expect(last_response).to be_ok + expect(JSON.parse(last_response.body).count { |entry| entry["domain"] == "duplicate.example" }).to eq(2) + + allow(application_class).to receive(:federation_target_domains).and_return([]) + application_class.announce_instance_to_all_domains + PotatoMesh::App::ApiCache.invalidate_prefix("api:instances:") + get "/api/instances" expect(last_response).to be_ok @@ -6860,8 +6869,9 @@ RSpec.describe "Potato Mesh Sinatra app" do ) end - # Prime the newest-page cache with just node A. - get "/api/nodes?limit=3" + # Prime the newest-page cache with just node A, at a limit the app + # requests: only those pages are cached (SPEC RC3). + get "/api/nodes?limit=10" expect(last_response).to be_ok expect(JSON.parse(last_response.body).map { |r| r["node_id"] }).to eq(["!cacheaaaa"]) @@ -6874,18 +6884,18 @@ RSpec.describe "Potato Mesh Sinatra app" do end # A plain request is still served the cached page (proves the cache is live). - get "/api/nodes?limit=3" + get "/api/nodes?limit=10" expect(JSON.parse(last_response.body).map { |r| r["node_id"] }).to eq(["!cacheaaaa"]) # A `before` request bypasses the cache and sees the fresh node B. - get "/api/nodes?limit=3&before=#{now}" + get "/api/nodes?limit=10&before=#{now}" expect(last_response).to be_ok before_ids = JSON.parse(last_response.body).map { |r| r["node_id"] } expect(before_ids).to include("!cacheaaaa", "!cachebbbb") # Issuing the `before` request did not overwrite the hot newest-page cache # entry — a subsequent plain request still returns the cached page. - get "/api/nodes?limit=3" + get "/api/nodes?limit=10" expect(JSON.parse(last_response.body).map { |r| r["node_id"] }).to eq(["!cacheaaaa"]) end @@ -7179,7 +7189,7 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(bucket["total"]).to eq(5400) # SUM across protocols end - it "bypasses the cache when since is provided" do + it "answers since from the canonical cached series (SPEC RC1)" do clear_database get "/api/stats/activity?window_seconds=86400&bucket_seconds=3600&since=#{Time.now.to_i - 3600}" expect(last_response).to be_ok @@ -7205,10 +7215,11 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(JSON.parse(last_response.body)).to eq("error" => "bucket_seconds must be positive") end - it "rejects a bucket too small for the requested window" do + it "serves a bucket too small for the requested window as the canonical pair (SPEC RC1)" do + clear_database get "/api/stats/activity?window_seconds=86400&bucket_seconds=1" - expect(last_response.status).to eq(400) - expect(JSON.parse(last_response.body)).to eq("error" => "bucket_seconds too small for requested window") + expect(last_response.status).to eq(200) + expect(JSON.parse(last_response.body)).to be_an(Array) end end @@ -8763,10 +8774,15 @@ RSpec.describe "Potato Mesh Sinatra app" do get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=0" expect(last_response.status).to eq(400) expect(JSON.parse(last_response.body)).to eq("error" => "bucketSeconds must be positive") + end + + it "serves a bucket too small for the requested window as the canonical pair (SPEC RC1)" do + post "/api/telemetry", telemetry_fixture.to_json, auth_headers + expect(last_response.status).to eq(201) get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=1" - expect(last_response.status).to eq(400) - expect(JSON.parse(last_response.body)).to eq("error" => "bucketSeconds too small for requested window") + expect(last_response.status).to eq(200) + expect(JSON.parse(last_response.body).map { |bucket| bucket["bucket_seconds"] }.uniq).to eq([300]) end it "clamps windowSeconds to the 28-day visibility cap" do diff --git a/web/spec/instances_spec.rb b/web/spec/instances_spec.rb index fedd96fe..cf370438 100644 --- a/web/spec/instances_spec.rb +++ b/web/spec/instances_spec.rb @@ -97,6 +97,21 @@ RSpec.describe PotatoMesh::App::Instances do expect(payload.all? { |row| row["last_update"] >= lower_bound }).to be(true) end + it "skips a row whose stored update time is not a number" do + # SQLite ranks TEXT above every INTEGER, so the SQL window lets the row + # through; the Ruby check after normalisation drops it. + with_db do |db| + db.execute( + "INSERT INTO instances (id, domain, pubkey, last_update_time, is_private) VALUES (?, ?, ?, ?, ?)", + ["text-time-instance", "text-time.mesh.test", PotatoMesh::Application::INSTANCE_PUBLIC_KEY_PEM, "soon", 0], + ) + end + + domains = application_class.load_instances_for_api.map { |row| row["domain"] } + + expect(domains).not_to include("text-time.mesh.test") + end + it "exposes contactLink when present and omits blank values" do fixed_time = Time.utc(2025, 2, 1, 12, 0, 0) allow(Time).to receive(:now).and_return(fixed_time) diff --git a/web/spec/private_hidden_client_spec.rb b/web/spec/private_hidden_client_spec.rb index fb3fb346..1bad89a3 100644 --- a/web/spec/private_hidden_client_spec.rb +++ b/web/spec/private_hidden_client_spec.rb @@ -256,7 +256,8 @@ RSpec.describe "Private mode and CLIENT_HIDDEN nodes" do end end - # Samples the telemetry rollup counted in the last hour. + # Samples the telemetry rollup counted in the last day: a one-hour request + # is served the canonical 1-day pair (SPEC RC1). # # @return [Integer] summed +sample_count+. def aggregated_samples diff --git a/web/spec/read_cache_keys_spec.rb b/web/spec/read_cache_keys_spec.rb new file mode 100644 index 00000000..381981db --- /dev/null +++ b/web/spec/read_cache_keys_spec.rb @@ -0,0 +1,499 @@ +# 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. + +# frozen_string_literal: true + +require "spec_helper" +require "sqlite3" +require "json" + +# The read path's cache keys and predicates (SPEC RC1-RC4, security and +# performance review): the bucketed endpoints serve canonical window/bucket +# pairs, every rx_time window searches its index, the response cache holds a +# bounded key set whose fixed entries no variant evicts, and a +# GET /api/instances miss only reads. +RSpec.describe "Read-path cache keys and predicates" do + let(:app) { Sinatra::Application } + let(:application_class) { PotatoMesh::Application } + # A fixed clock 123 s past a bucket boundary of every canonical width. + let(:reference_time) { Time.at(1_791_000_123) } + let(:now) { reference_time.to_i } + + # Execute the block with a read-write connection to the spec database. + # + # @param readonly [Boolean] whether to open the database read-only. + # @yieldparam db [SQLite3::Database] open handle. + # @return [Object] the block's value. + def with_db(readonly: false) + db = SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: readonly) + db.busy_timeout = PotatoMesh::Config.db_busy_timeout_ms + yield db + ensure + db&.close + end + + # Empty every table the examples read. + # + # @return [void] + def clear_tables + with_db do |db| + %w[instances trace_hops traces neighbors messages nodes positions telemetry ingestors ingestor_activity waypoints].each do |table| + db.execute("DELETE FROM #{table}") + end + end + end + + # GET +path+ and parse its JSON body. + # + # @param path [String] request path with query. + # @return [Object] parsed body. + def get_json(path) + get path + expect(last_response.status).to eq(200), "GET #{path} answered #{last_response.status}" + JSON.parse(last_response.body) + end + + # Store one telemetry reading straight in the database, past every cache. + # + # @param id [Integer] row id. + # @param rx_time [Integer] receive time. + # @param battery [Float] battery level. + # @return [void] + def insert_telemetry(id, rx_time, battery: 50.0) + with_db do |db| + db.execute( + "INSERT INTO telemetry(id, node_id, rx_time, rx_iso, battery_level) VALUES(?,?,?,?,?)", + [id, "!rc000001", rx_time, Time.at(rx_time).utc.iso8601, battery], + ) + end + end + + # Store one ingestor activity row straight in the database. + # + # @param at [Integer] activity timestamp. + # @param packets [Integer] packets counted. + # @return [void] + def insert_activity(at, packets: 3600) + with_db do |db| + db.execute( + "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)", + ["!rc0000aa", at, packets, "meshtastic"], + ) + end + end + + # Store one node heard at +last_heard+ straight in the database. + # + # @param node_id [String] canonical node id. + # @param num [Integer] node number. + # @param last_heard [Integer] last heard time. + # @return [void] + def insert_node(node_id, num, last_heard) + with_db do |db| + db.execute( + "INSERT INTO nodes(node_id, num, short_name, long_name, last_heard, first_heard) VALUES(?,?,?,?,?,?)", + [node_id, num, "RC#{num % 100}", "Read cache #{num}", last_heard, last_heard], + ) + end + end + + # Count the SQL statements matching +pattern+ that the block issues. + # + # @param pattern [Regexp] statement pattern. + # @yield the requests to count. + # @return [Integer] matching statements executed. + def count_statements(pattern) + count = 0 + allow_any_instance_of(SQLite3::Database).to receive(:execute).and_wrap_original do |original, sql, *args, &block| + count += 1 if sql.is_a?(String) && sql.match?(pattern) + original.call(sql, *args, &block) + end + yield + count + end + + # The bucket widths of a series, from its +bucket_start+ and +bucket_end+. + # + # @param series [Array] parsed buckets. + # @return [Array] distinct widths. + def bucket_widths(series) + series.map { |bucket| bucket["bucket_end"] - bucket["bucket_start"] }.uniq + end + + before do + allow(Time).to receive(:now).and_return(reference_time) + ENV.delete("PRIVATE") + clear_tables + PotatoMesh::App::ApiCache.invalidate_all + end + + after { PotatoMesh::App::ApiCache.invalidate_all } + + # Statement shapes of the two bucketed queries. + TELEMETRY_BUCKETS_SQL = /AS sample_count/ + ACTIVITY_BUCKETS_SQL = /ingestor_total/ + + describe "canonical windows of the bucketed endpoints (RC1)" do + it "computes /api/telemetry/aggregated at most once per canonical pair for eleven distinct windows" do + insert_telemetry(1, now - 120) + windows = [3_600, 7_200, 43_200, 86_400, 172_800, 259_200, 345_601, 432_000, 604_800, 1_209_600, 2_419_200] + computed = count_statements(TELEMETRY_BUCKETS_SQL) do + windows.each { |window| get_json("/api/telemetry/aggregated?windowSeconds=#{window}&bucketSeconds=3600") } + end + # Two canonical pairs: 1 day in 5-minute and 7 days in 1-hour buckets. + expect(computed).to be <= 2 + end + + it "computes /api/stats/activity at most once per canonical pair for eleven distinct windows" do + insert_activity(now - 120) + windows = [3_600, 7_200, 43_200, 86_400, 172_800, 259_200, 345_601, 432_000, 604_800, 1_209_600, 2_419_200] + computed = count_statements(ACTIVITY_BUCKETS_SQL) do + windows.each { |window| get_json("/api/stats/activity?window_seconds=#{window}&bucket_seconds=3600") } + end + # Two canonical pairs: 1 day in 1-hour and 7 days in 2-hour buckets. + expect(computed).to be <= 2 + end + + it "serves a telemetry request the nearest canonical pair and reports its bucket" do + insert_telemetry(1, now - 5 * 3600) + insert_telemetry(2, now - 2 * 86_400) + insert_telemetry(3, now - 120) + + # One hour in one-minute buckets: the 1-day pair, both readings of the day. + day = get_json("/api/telemetry/aggregated?windowSeconds=3600&bucketSeconds=60") + expect(day.map { |bucket| bucket["bucket_seconds"] }.uniq).to eq([300]) + expect(day.sum { |bucket| bucket["sample_count"] }).to eq(2) + + # 28 days in 1-day buckets: the 7-day pair, all three readings. + week = get_json("/api/telemetry/aggregated?windowSeconds=2419200&bucketSeconds=86400") + expect(week.map { |bucket| bucket["bucket_seconds"] }.uniq).to eq([3600]) + expect(week.sum { |bucket| bucket["sample_count"] }).to eq(3) + + # The window decides before the bucket: 7 days in 20-minute buckets is the 7-day pair. + expect(get_json("/api/telemetry/aggregated?windowSeconds=604800&bucketSeconds=1200").map { |b| b["bucket_seconds"] }.uniq).to eq([3600]) + # One second past the midpoint of the two windows is nearer the week. + expect(get_json("/api/telemetry/aggregated?windowSeconds=345601&bucketSeconds=600").map { |b| b["bucket_seconds"] }.uniq).to eq([3600]) + # Equally near both pairs in window and bucket: the shorter window. + expect(get_json("/api/telemetry/aggregated?windowSeconds=345600&bucketSeconds=1950").map { |b| b["bucket_seconds"] }.uniq).to eq([300]) + end + + it "serves an activity request the nearest canonical pair" do + insert_activity(now - 5 * 3600) + insert_activity(now - 2 * 86_400) + insert_activity(now - 120) + + day = get_json("/api/stats/activity?window_seconds=3600&bucket_seconds=600") + expect(bucket_widths(day)).to eq([3600]) + expect(day.length).to eq(2) + + week = get_json("/api/stats/activity?window_seconds=2419200&bucket_seconds=86400") + expect(bucket_widths(week)).to eq([7200]) + expect(week.length).to eq(3) + + expect(bucket_widths(get_json("/api/stats/activity?window_seconds=604800&bucket_seconds=3600"))).to eq([7200]) + expect(bucket_widths(get_json("/api/stats/activity?window_seconds=345600&bucket_seconds=5400"))).to eq([3600]) + end + + it "rounds a telemetry since down to its bucket and answers it from the canonical entry" do + boundary = (now / 300) * 300 - 3000 + insert_telemetry(1, boundary + 10, battery: 80.0) + + computed = count_statements(TELEMETRY_BUCKETS_SQL) do + get_json("/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300") + # A since inside the bucket keeps the whole bucket, the reading included. + series = get_json("/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300&since=#{boundary + 200}") + expect(series.map { |bucket| bucket["bucket_start"] }).to eq([boundary]) + expect(series.first.dig("aggregates", "battery_level", "avg")).to eq(80.0) + # The next bucket on is empty. + expect(get_json("/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300&since=#{boundary + 300}")).to eq([]) + 20.times { |i| get_json("/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300&since=#{now - 7200 + i * 37}") } + end + expect(computed).to eq(1) + end + + it "rounds an activity since down to its bucket and answers it from the canonical entry" do + boundary = (now / 3600) * 3600 - 7200 + insert_activity(boundary + 10) + + computed = count_statements(ACTIVITY_BUCKETS_SQL) do + get_json("/api/stats/activity") + series = get_json("/api/stats/activity?since=#{boundary + 1800}") + expect(series.map { |bucket| bucket["bucket_start"] }).to eq([boundary]) + expect(get_json("/api/stats/activity?since=#{boundary + 3600}")).to eq([]) + 20.times { |i| get_json("/api/stats/activity?since=#{now - 7200 + i * 37}") } + end + expect(computed).to eq(1) + end + + it "answers a since at or before the window floor with the canonical body" do + insert_telemetry(1, now - 120) + insert_activity(now - 120) + + get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300" + body = last_response.body + etag = last_response.headers["ETag"] + get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300&since=1" + expect(last_response.body).to eq(body) + expect(last_response.headers["ETag"]).to eq(etag) + + get "/api/stats/activity" + activity = last_response.body + get "/api/stats/activity?since=#{now - 86_400}" + expect(last_response.body).to eq(activity) + end + + it "serves a bucket too small for its window as the canonical pair" do + insert_telemetry(1, now - 120) + insert_activity(now - 120) + + expect(get_json("/api/telemetry/aggregated?windowSeconds=604800&bucketSeconds=300").map { |b| b["bucket_seconds"] }.uniq).to eq([3600]) + expect(get_json("/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=1").map { |b| b["bucket_seconds"] }.uniq).to eq([300]) + expect(bucket_widths(get_json("/api/stats/activity?window_seconds=86400&bucket_seconds=1"))).to eq([3600]) + end + + it "keeps answering a non-positive window or bucket with 400" do + get "/api/telemetry/aggregated?windowSeconds=0&bucketSeconds=300" + expect(last_response.status).to eq(400) + get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=-300" + expect(last_response.status).to eq(400) + get "/api/stats/activity?window_seconds=-1&bucket_seconds=3600" + expect(last_response.status).to eq(400) + get "/api/stats/activity?window_seconds=86400&bucket_seconds=0" + expect(last_response.status).to eq(400) + end + end + + describe "rx_time predicates (RC2)" do + # Run the block, capture the first statement it issues against +table+ + # and return the detail lines of that statement's query plan. + # + # @param table [String] table the statement reads. + # @yield the query helper call. + # @return [Array] EXPLAIN QUERY PLAN detail lines. + def plan_for(table) + captured = nil + allow_any_instance_of(SQLite3::Database).to receive(:execute).and_wrap_original do |original, sql, *args, &block| + captured ||= [sql, args.first] if sql.is_a?(String) && sql.match?(/FROM #{table}\b/) + original.call(sql, *args, &block) + end + yield + sql, binds = captured + expect(sql).not_to be_nil, "no statement read #{table}" + with_db(readonly: true) do |db| + db.execute("EXPLAIN QUERY PLAN #{sql}", binds || []).map { |row| row[3] } + end + end + + # A plan reading +table+ through its rx_time index with +bounds+. + # + # @param table [String] table name. + # @param bounds [String] the index constraint as EXPLAIN prints it. + # @return [RSpec::Matchers::BuiltIn::Include] matcher. + def search_rx_time(table, bounds) + include(a_string_matching(/\ASEARCH (TABLE )?#{table} USING INDEX idx_#{table}_rx_time \(#{Regexp.escape(bounds)}\)/)) + end + + { + "telemetry" => ->(before) { PotatoMesh::Application.query_telemetry(1000, before: before) }, + "positions" => ->(before) { PotatoMesh::Application.query_positions(1000, before: before) }, + "neighbors" => ->(before) { PotatoMesh::Application.query_neighbors(1000, before: before) }, + "traces" => ->(before) { PotatoMesh::Application.query_traces(1000, before: before) }, + }.each do |table, query| + it "searches idx_#{table}_rx_time for the #{table} window" do + plan = plan_for(table) { query.call(nil) } + expect(plan).to search_rx_time(table, "rx_time>?") + expect(plan.grep(/\ASCAN (TABLE )?#{table}\b/)).to eq([]) + end + + it "searches idx_#{table}_rx_time between the window floor and before" do + plan = plan_for(table) { query.call(now - 60) } + expect(plan).to search_rx_time(table, "rx_time>? AND rx_time?") + expect(plan.grep(/\ASCAN (TABLE )?telemetry\b/)).to eq([]) + end + end + + describe "bounded response cache keys (RC3)" do + it "keeps /api/stats cached through 70 distinct /api/nodes limits" do + insert_node("!rc0a0001", 0x0a0001, now - 60) + cached = get_json("/api/stats")["total"]["nodes"]["day"] + # Written past the ingest routes, so no invalidation: only a recompute shows it. + insert_node("!rc0a0002", 0x0a0002, now - 60) + + 70.times { |i| get_json("/api/nodes?limit=#{i + 1}") } + + expect(get_json("/api/stats")["total"]["nodes"]["day"]).to eq(cached) + end + + it "keeps the fixed entries through more protocol variants than the cache holds" do + insert_node("!rc0a0001", 0x0a0001, now - 60) + stats = get_json("/api/stats")["total"]["nodes"]["day"] + page = get_json("/api/nodes?limit=1000").length + insert_node("!rc0a0002", 0x0a0002, now - 60) + + routes = %w[/api/nodes /api/positions /api/telemetry /api/neighbors /api/traces /api/ingestors /api/waypoints /api/messages] + variants = routes.product(%w[meshcore meshtastic reticulum], [10, 200, 1000]) + expect(variants.length).to be > PotatoMesh::App::ApiCache::MAX_ENTRIES + variants.each { |path, protocol, limit| get_json("#{path}?limit=#{limit}&protocol=#{protocol}") } + + expect(get_json("/api/stats")["total"]["nodes"]["day"]).to eq(stats) + expect(get_json("/api/nodes?limit=1000").length).to eq(page) + end + + it "serves a limit outside the cached set fresh and caches the limits the app requests" do + insert_node("!rc0a0001", 0x0a0001, now - 60) + expect(get_json("/api/nodes?limit=7").length).to eq(1) + expect(get_json("/api/nodes?limit=10").length).to eq(1) + insert_node("!rc0a0002", 0x0a0002, now - 60) + + expect(get_json("/api/nodes?limit=7").length).to eq(2) + expect(get_json("/api/nodes?limit=10").length).to eq(1) + end + end + + describe "GET /api/instances reads only (RC4)" do + # Two rows naming one domain in different forms, the older first, and + # a self record an hour stale. + # + # @return [void] + def seed_duplicates_and_stale_self! + application_class.ensure_self_instance_record! + with_db do |db| + sql = "INSERT INTO instances (id, domain, pubkey, name, last_update_time, is_private, signature) VALUES (?, ?, ?, ?, ?, 0, ?)" + pem = OpenSSL::PKey::RSA.new(2048).public_key.export + db.execute(sql, ["rc-dup-old", "rc-dup.example ", pem, "Old", now - 600, "sig-old"]) + db.execute(sql, ["rc-dup-new", "RC-Dup.Example", pem, "New", now - 60, "sig-new"]) + db.execute("UPDATE instances SET last_update_time = ? WHERE id = ?", [now - 3600, SELF_INSTANCE_ID]) + end + end + + # Every stored instance row (id prefix, domain, name, last update, + # signature prefix), ordered by id. + # + # @return [Array] rows. + def instance_rows + with_db(readonly: true) do |db| + db.execute("SELECT substr(id, 1, 10), domain, name, last_update_time, substr(signature, 1, 8) FROM instances ORDER BY id") + end + end + + it "answers a cache miss without changing a stored row" do + seed_duplicates_and_stale_self! + before_rows = instance_rows + + ids = get_json("/api/instances").map { |entry| entry["id"] } + + expect(ids).to include(SELF_INSTANCE_ID, "rc-dup-new") + expect(instance_rows).to eq(before_rows) + end + + # Our own row as the last cycle left it, older than the 7-day window, + # and one node heard at +last_heard+. + # + # @param last_heard [Integer] the node's last heard time. + # @return [void] + def stale_self_with_node!(last_heard) + application_class.ensure_self_instance_record! + with_db do |db| + db.execute( + "UPDATE instances SET last_update_time = ? WHERE id = ?", + [now - PotatoMesh::Config.week_seconds - 3600, SELF_INSTANCE_ID], + ) + end + insert_node("!rc0b0001", 0x0b0001, last_heard) + end + + it "lists our own row once a node is heard again, before the next cycle" do + stale_self_with_node!(now - 60) + before_rows = instance_rows + + ids = get_json("/api/instances").map { |entry| entry["id"] } + + expect(ids).to include(SELF_INSTANCE_ID) + expect(instance_rows).to eq(before_rows) + end + + it "leaves our own row out while its newest node is older than the window" do + stale_self_with_node!(now - PotatoMesh::Config.week_seconds - 60) + + expect(get_json("/api/instances").map { |entry| entry["id"] }).not_to include(SELF_INSTANCE_ID) + end + + it "deduplicates and refreshes the self record on the federation cycle" do + seed_duplicates_and_stale_self! + allow(application_class).to receive(:federation_target_domains).and_return([]) + + application_class.announce_instance_to_all_domains + + rows = with_db(readonly: true) { |db| db.execute("SELECT id, domain, last_update_time FROM instances ORDER BY id") } + expect(rows.map(&:first)).to contain_exactly("rc-dup-new", SELF_INSTANCE_ID) + expect(rows.to_h { |id, domain, _| [id, domain] }["rc-dup-new"]).to eq("rc-dup.example") + expect(rows.to_h { |id, _, updated| [id, updated] }[SELF_INSTANCE_ID]).to eq(now) + end + end + + describe "cost parameters from the query string only (RC6)" do + # GET +path+ with a form-encoded body and parse the JSON answer. + # + # @param path [String] request path with query. + # @param body [String] form-encoded request body. + # @return [Object] parsed body. + def get_json_with_form_body(path, body) + get path, {}, { input: body, "CONTENT_TYPE" => "application/x-www-form-urlencoded" } + expect(last_response.status).to eq(200), "GET #{path} with body #{body} answered #{last_response.status}" + JSON.parse(last_response.body) + end + + it "answers a GET whose form body carries since, before or limit as if the body were absent" do + insert_node("!rc0c0001", 0x0c0001, now - 3600) + insert_node("!rc0c0002", 0x0c0002, now - 10) + # limit=7 is served uncached (RC3), so every answer is a fresh query. + both = get_json("/api/nodes?limit=7").map { |node| node["node_id"] }.sort + expect(both.length).to eq(2) + + ["since=#{now - 30}", "before=#{now - 30}", "limit=1"].each do |body| + ids = get_json_with_form_body("/api/nodes?limit=7", body).map { |node| node["node_id"] }.sort + expect(ids).to eq(both), "a body of #{body} changed the answer" + end + end + + it "takes the bucketed endpoints' window and bucket from the query string only" do + insert_telemetry(1, now - 120) + insert_activity(now - 120) + + series = get_json_with_form_body("/api/telemetry/aggregated", "windowSeconds=604800&bucketSeconds=3600") + expect(series.map { |bucket| bucket["bucket_seconds"] }.uniq).to eq([300]) + activity = get_json_with_form_body("/api/stats/activity", "window_seconds=604800&bucket_seconds=7200") + expect(bucket_widths(activity)).to eq([3600]) + end + + it "keeps a cached page cached when a form body carries since, before or limit" do + insert_node("!rc0c0001", 0x0c0001, now - 60) + expect(get_json("/api/nodes?limit=10").length).to eq(1) + # Written past the ingest routes, so only a fresh query shows it. + insert_node("!rc0c0002", 0x0c0002, now - 30) + + ["since=1", "before=#{now}", "limit=999"].each do |body| + expect(get_json_with_form_body("/api/nodes?limit=10", body).length).to eq(1), "a body of #{body} bypassed the cache" + end + end + end +end