web,data: cap ingested strings by column and check node ids (#997)

data/mesh_ingestor/CONTRACTS.md

@@ -14,6 +14,7 @@ This document records the contracts that future protocols must preserve. New pro
 
 - Canonical node id: `nodes.node_id` is a `TEXT` primary key and is treated as canonical across the system.
 - Format: `!%08x` (lowercase hex, 8 chars), for example `!abcdef01`.
+- Ingest id forms (SPEC SL5): a node reference (`from_id`, `node_id`, `neighbor_id`, a heartbeat's `node_id`) is the canonical id, a node number (a JSON integer from 0 to 4294967295, or its decimal string), or absent; a destination (`to_id`) may also be the broadcast id `^all`. The `ingestor` field and the keys of `POST /api/nodes` take the canonical id only. The web app skips a record carrying any other id, and a neighbour entry carrying one, and still answers 201; a heartbeat with one answers 400.
 - Normalization:
   - Python currently normalizes via `data/mesh_ingestor/serialization.py:_canonical_node_id`.
   - Ruby normalizes via `web/lib/potato_mesh/application/data_processing.rb:canonical_node_parts`.
@@ -126,6 +127,8 @@ Payload is a mapping keyed by canonical node id, with optional top-level `”ing
 
 - `{ “!abcdef01”: { ... node fields ... }, “ingestor”: “!ingestornodeid”, “protocol”: “meshcore” }`
 
+A key that is not a canonical node id, or whose entry is not a mapping, is skipped, and the request still answers 201 (SPEC SL5). A nested `user`, `deviceMetrics`, `device_metrics`, `position`, `position.raw` or `destination` that is not a mapping is ignored, as if absent (SPEC SL10).
+
 Protocol resolution per-row honours, in order: (1) an explicit per-node `”protocol”` field inside the node entry; (2) the wrapper-level top-level `”protocol”` key; (3) the registered ingestor's protocol (see `POST /api/ingestors`); (4) `”meshtastic”` as the final default. Valid values are `”meshtastic”`, `”meshcore”`, and `”reticulum”` - values outside this set fall through to the next source. The wrapper stamp is what the Python ingestor emits unconditionally so the web app classifies records correctly even before the ingestor heartbeat is processed (closes the startup race that misclassified MeshCore placeholders as Meshtastic).
 
 Node entry fields are “Meshtastic-ish” (camelCase) and may include the following.
@@ -350,11 +353,11 @@ the opt-out marker are excluded from every read surface.
 
 Heartbeat payload:
 
-- `node_id` (canonical string)
+- `node_id` (canonical string; a value that is not a node reference answers `400`)
 - `start_time` (int), `last_seen_time` (int)
 - `version` (string)
 - Optional: `lora_freq`, `modem_preset`
-- Optional: `protocol` (string; e.g. `"meshtastic"`, `"meshcore"`, `"reticulum"`) - declares the mesh backend for this ingestor; defaults to `"meshtastic"` when absent
+- Optional: `protocol` (string; `"meshtastic"`, `"meshcore"` or `"reticulum"`, case and surrounding spaces ignored) - declares the mesh backend for this ingestor; defaults to `"meshtastic"` when absent or any other value (SPEC SL7)
 - Optional: `packets` (int ≥ 0) - mesh-activity delta (SPEC MA1/MA2). The merged count of *every* frame this ingestor handled since its previous heartbeat: all received frames (including ignored / errored / unimplemented) plus its own transmissions (announcement + MeshCore telemetry polls), counted at the earliest receive/transmit seam so nothing is under-reported. It is a per-interval delta (reset on each send), not a since-boot cumulative. Additive and backward-compatible: an absent or negative value records no activity, so pre-feature ingestors are unaffected.
 
 Mesh-activity time-series (SPEC MA3). Each heartbeat carrying a non-negative `packets` value appends one append-only row to the `ingestor_activity` table (`ingestor_id`, `at`, `packets`, `protocol`; `data/ingestor_activity.sql`); the `ingestors` snapshot row is upserted as before. Each ingestor's contribution is stored separately (never pre-summed) so a packets/hour moving average is computable across time × protocol × multiple ingestors. The row is best-effort - a failed activity insert never sinks the liveness heartbeat (still `201`). Rows are pruned by the retention worker on `at`. The read-side aggregate is served by `GET /api/stats` (`<scope>.packets.hour`, below).
@@ -363,6 +366,36 @@ Protocol propagation: all event records (`messages`, `positions`, `telemetry`, `
 
 POST response & validation (0.7.0). Every `POST /api/*` ingest route returns `201 Created` with `{"status":"ok"}` on success (`POST /api/instances` returns `{"status":"registered"}`). A batch route (`messages` / `positions` / `telemetry` / `neighbors` / `traces`) accepts either a single record object or an array of them; any other top-level JSON type is rejected with `400 {"error":"invalid payload"}`, matching the `/api/nodes` and `/api/ingestors` object check. Clients should treat any `2xx` as success.
 
+Field limits (SPEC SL1-SL10). Every ingest write bounds the strings it stores, in UTF-8 bytes, and still answers 201. Free text (T) is cut to its longest prefix of whole grapheme clusters within the cap (whole code points when one cluster alone is longer). A token (N: an id, key, enum label or encoded payload) over its cap is stored as `NULL`. A value within its cap is stored as posted. A numeric field is stored as a number or `NULL`, never as text: a numeric string such as `"5.5"` is converted, and text that is no number, a mapping or a list is stored as `NULL` (node `hopsAway`, `snr`, `rssi`, `isFavorite`, `user.isUnmessagable`, `deviceMetrics.*` and `position.altitude`; message `channel`, `snr`, `rssi` and `hop_limit`; every other numeric field was already converted). In every numeric field, an integer outside the signed 64-bit range and a number that is not finite are stored as `NULL`. Rows stored before the caps existed are left as they are and age out under retention.
+
+| Field | Cap (bytes) | Policy |
+| --- | --- | --- |
+| node `user.longName`, a destination's name | 512 | T |
+| node `user.shortName` | 16 | T |
+| node `user.hwModel` / `hwModel` | 64 | N |
+| node `user.role`, `destination.role` | 32 | N |
+| node `user.macaddr` | 32 | N |
+| node `user.publicKey` | 512 | N; a key stored as `NULL` is no keyed evidence (SPEC SL4) |
+| node `identityHash`, `destination.id` | 64 | N; a destination without an id is not stored |
+| `destination.aspect` | 64 | N |
+| node `interface` | 256 | T |
+| `location_source`, `modem_preset`, `portnum`, `telemetry_type` | 32 | N |
+| message `text` | 1024 | T |
+| message `encrypted`, every `payload_b64` | 512 | N |
+| message `channel_name` | 64 | T |
+| message `emoji` | 64 | N |
+| message `path` | 512 | N |
+| every `rx_iso` | 32 | N; the web app derives a missing one from `rx_time` |
+| telemetry `user_string` | 256 | T, wherever the record nests it |
+| telemetry `one_wire_temperature` | 8 entries | the first 8 are kept |
+| waypoint `name` | 128 | T |
+| waypoint `description` | 512 | T |
+| heartbeat `version` | 64 | T |
+
+Each cap sits well above the longest value the radio protocols produce. The ingestor trims a string longer than its cap plus 64 bytes, on a code-point boundary, before it posts it (`data/mesh_ingestor/field_limits.py`, SPEC SL8); the web app makes the final cut (`web/lib/potato_mesh/application/data_processing/field_limits.rb`). A field added to a payload gets a cap in both tables.
+
+A signed instance field is never cut, since a cut value no longer matches its signature (SPEC SL6). `POST /api/instances` answers `400 {"error":"name exceeds 256 bytes"}` (likewise `version`, `channel`, `frequency` and `contact_link`), `400 {"error":"public_key exceeds 2048 bytes"}` and `400 {"error":"signature exceeds 1024 bytes"}` before it checks the signature; a crawl skips such a record from a peer's `/api/instances` and logs `warn` "Discarded remote instance entry" with that reason.
+
 ### GET endpoint filtering
 
 All collection GET endpoints (`/api/nodes`, `/api/messages`, `/api/positions`, `/api/telemetry`, `/api/traces`, `/api/neighbors`, `/api/ingestors`, `/api/waypoints`) accept an optional `?protocol=<value>` query parameter. When present, only records whose `protocol` column matches the given value are returned. The `protocol` field is included in all GET responses.

data/mesh_ingestor/field_limits.py

@@ -0,0 +1,253 @@
+# 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.
+
+"""Loose byte caps on the strings the ingestor posts (SPEC SL8).
+
+The web app holds the final caps, in UTF-8 bytes, and cuts free text on a
+grapheme-cluster boundary
+(``web/lib/potato_mesh/application/data_processing/field_limits.rb``).
+Python's standard library has no grapheme segmentation, so the ingestor only
+trims a string that runs past its web cap plus :data:`LOOSE_MARGIN_BYTES`, on a
+code-point boundary.  The web's cut reads the same margin past its cap, so it
+sees in a trimmed value the context it would see in the untrimmed one and
+stores the same text; the trim only keeps an oversized value off the wire.  A
+protobuf parses a string of any length, whatever the firmware allows, so a
+node name can arrive at 60 kB.
+
+:func:`bound_post_payload` runs once per payload, in
+:func:`~data.mesh_ingestor.queue._queue_post_json`, the one place every POST
+to the web app passes.
+"""
+
+from __future__ import annotations
+
+from typing import Final, Mapping
+
+LOOSE_MARGIN_BYTES: Final[int] = 64
+"""Bytes kept past a web cap; the web's cut reads the same margin."""
+
+CAPS: Final[Mapping[str, int]] = {
+    "LONG_NAME": 512,
+    "SHORT_NAME": 16,
+    "HW_MODEL": 64,
+    "ROLE": 32,
+    "MACADDR": 32,
+    "PUBLIC_KEY": 512,
+    "HASH": 64,
+    "ASPECT": 64,
+    "INTERFACE": 256,
+    "LABEL": 32,
+    "MESSAGE_TEXT": 1024,
+    "PAYLOAD": 512,
+    "CHANNEL_NAME": 64,
+    "EMOJI": 64,
+    "PATH": 512,
+    "RX_ISO": 32,
+    "USER_STRING": 256,
+    "WAYPOINT_NAME": 128,
+    "WAYPOINT_DESCRIPTION": 512,
+    "INGESTOR_VERSION": 64,
+}
+"""Web caps in UTF-8 bytes, named as the web's ``<NAME>_BYTES`` constants."""
+
+_NODE_FIELDS: Final[Mapping[tuple[str, ...], int]] = {
+    ("user", "longName"): CAPS["LONG_NAME"],
+    ("user", "long_name"): CAPS["LONG_NAME"],
+    ("user", "shortName"): CAPS["SHORT_NAME"],
+    ("user", "short_name"): CAPS["SHORT_NAME"],
+    ("user", "hwModel"): CAPS["HW_MODEL"],
+    ("user", "hw_model"): CAPS["HW_MODEL"],
+    ("hwModel",): CAPS["HW_MODEL"],
+    ("hw_model",): CAPS["HW_MODEL"],
+    ("user", "role"): CAPS["ROLE"],
+    ("user", "macaddr"): CAPS["MACADDR"],
+    ("user", "publicKey"): CAPS["PUBLIC_KEY"],
+    ("user", "public_key"): CAPS["PUBLIC_KEY"],
+    ("identityHash",): CAPS["HASH"],
+    ("identity_hash",): CAPS["HASH"],
+    ("destination", "id"): CAPS["HASH"],
+    ("destination", "aspect"): CAPS["ASPECT"],
+    ("destination", "role"): CAPS["ROLE"],
+    ("interface",): CAPS["INTERFACE"],
+    ("modem_preset",): CAPS["LABEL"],
+    ("modemPreset",): CAPS["LABEL"],
+    ("position", "locationSource"): CAPS["LABEL"],
+    ("position", "location_source"): CAPS["LABEL"],
+}
+
+ROUTE_FIELDS: Final[Mapping[str, Mapping[tuple[str, ...], int]]] = {
+    "/api/nodes": _NODE_FIELDS,
+    "/api/messages": {
+        ("text",): CAPS["MESSAGE_TEXT"],
+        ("encrypted",): CAPS["PAYLOAD"],
+        ("channel_name",): CAPS["CHANNEL_NAME"],
+        ("emoji",): CAPS["EMOJI"],
+        ("path",): CAPS["PATH"],
+        ("rx_iso",): CAPS["RX_ISO"],
+        ("portnum",): CAPS["LABEL"],
+        ("modem_preset",): CAPS["LABEL"],
+    },
+    "/api/positions": {
+        ("rx_iso",): CAPS["RX_ISO"],
+        ("location_source",): CAPS["LABEL"],
+        ("position", "location_source"): CAPS["LABEL"],
+        ("position", "locationSource"): CAPS["LABEL"],
+        ("position", "raw", "location_source"): CAPS["LABEL"],
+        ("payload_b64",): CAPS["PAYLOAD"],
+        ("modem_preset",): CAPS["LABEL"],
+    },
+    "/api/telemetry": {
+        ("rx_iso",): CAPS["RX_ISO"],
+        ("portnum",): CAPS["LABEL"],
+        ("payload_b64",): CAPS["PAYLOAD"],
+        ("modem_preset",): CAPS["LABEL"],
+        ("telemetry_type",): CAPS["LABEL"],
+        ("user_string",): CAPS["USER_STRING"],
+        ("host_metrics", "user_string"): CAPS["USER_STRING"],
+        ("host_metrics", "userString"): CAPS["USER_STRING"],
+    },
+    "/api/traces": {("rx_iso",): CAPS["RX_ISO"]},
+    "/api/waypoints": {
+        ("rx_iso",): CAPS["RX_ISO"],
+        ("name",): CAPS["WAYPOINT_NAME"],
+        ("description",): CAPS["WAYPOINT_DESCRIPTION"],
+        ("payload_b64",): CAPS["PAYLOAD"],
+    },
+    "/api/ingestors": {
+        ("version",): CAPS["INGESTOR_VERSION"],
+        ("modem_preset",): CAPS["LABEL"],
+    },
+}
+"""Capped string fields of each POST route, as nested key paths."""
+
+_KEYED_ROUTES: Final[frozenset[str]] = frozenset({"/api/nodes"})
+"""Routes whose payload maps a node id to each record."""
+
+
+def loose_cut(text: str, cap: int) -> str:
+    """Trim ``text`` to its web cap plus :data:`LOOSE_MARGIN_BYTES`.
+
+    The cut lands on a code-point boundary and may split a grapheme cluster;
+    the web app makes the final cut on a cluster boundary.  A lone surrogate
+    in a trimmed value is dropped, so the result always encodes as UTF-8.
+
+    Parameters:
+        text: String value of a capped field.
+        cap: Web cap of the field in UTF-8 bytes.
+
+    Returns:
+        ``text`` itself when it fits ``cap + LOOSE_MARGIN_BYTES`` bytes,
+        otherwise its longest prefix of whole code points that does.
+    """
+
+    limit = cap + LOOSE_MARGIN_BYTES
+    # A code point takes at most four UTF-8 bytes: a string this short fits
+    # without encoding it.
+    if len(text) * 4 <= limit:
+        return text
+    encoded = text.encode("utf-8", "surrogatepass")
+    if len(encoded) <= limit:
+        return text
+    return encoded[:limit].decode("utf-8", "ignore")
+
+
+def _bound_path(value: object, path: tuple[str, ...], cap: int) -> object:
+    """Trim the string at ``path`` inside ``value``, copying what changes.
+
+    Parameters:
+        value: Mapping at the current level, or any other value.
+        path: Remaining keys, outermost first.
+        cap: Web cap of the field in UTF-8 bytes.
+
+    Returns:
+        ``value`` itself when nothing at ``path`` needs a trim, otherwise a
+        shallow copy of each level down to the trimmed string.
+    """
+
+    if not isinstance(value, dict) or path[0] not in value:
+        return value
+    current = value[path[0]]
+    if len(path) > 1:
+        bounded = _bound_path(current, path[1:], cap)
+    elif isinstance(current, str):
+        bounded = loose_cut(current, cap)
+    else:
+        bounded = current
+    if bounded is current:
+        return value
+    copy = dict(value)
+    copy[path[0]] = bounded
+    return copy
+
+
+def _bound_record(record: object, fields: Mapping[tuple[str, ...], int]) -> object:
+    """Trim every capped field of one record.
+
+    Parameters:
+        record: One record of a payload.
+        fields: Capped field paths of the route, with their caps.
+
+    Returns:
+        ``record`` itself when every field fits, otherwise a trimmed copy.
+    """
+
+    bounded = record
+    for path, cap in fields.items():
+        bounded = _bound_path(bounded, path, cap)
+    return bounded
+
+
+def bound_post_payload(path: str, payload: object) -> object:
+    """Trim the oversized strings of a payload bound for ``path``.
+
+    The payload is never modified: a payload whose fields all fit comes back
+    as the same object, and one with an oversized field comes back as a copy
+    sharing everything the trim did not touch.
+
+    Parameters:
+        path: Web API route the payload is posted to.
+        payload: JSON-serialisable body: one record, a list of records, or,
+            for ``/api/nodes``, a mapping of node ids to records.
+
+    Returns:
+        The payload, trimmed.
+    """
+
+    fields = ROUTE_FIELDS.get(path)
+    if fields is None:
+        return payload
+    if isinstance(payload, list):
+        records = [_bound_record(record, fields) for record in payload]
+        if all(new is old for new, old in zip(records, payload)):
+            return payload
+        return records
+    if path in _KEYED_ROUTES and isinstance(payload, dict):
+        bounded = payload
+        for key, record in payload.items():
+            trimmed = _bound_record(record, fields)
+            if trimmed is not record:
+                if bounded is payload:
+                    bounded = dict(payload)
+                bounded[key] = trimmed
+        return bounded
+    return _bound_record(payload, fields)
+
+
+__all__ = [
+    "CAPS",
+    "LOOSE_MARGIN_BYTES",
+    "ROUTE_FIELDS",
+    "bound_post_payload",
+    "loose_cut",
+]

data/mesh_ingestor/queue.py

@@ -24,7 +24,7 @@ import urllib.request
 from dataclasses import dataclass, field
 from typing import Callable, Iterable, Mapping, Tuple
 
-from . import config
+from . import config, field_limits
 
 
 def _stringify_payload_value(value: object) -> str:
@@ -481,6 +481,11 @@ def _queue_post_json(
     ``send`` override via :func:`_fresh_state`) and for any standalone use
     without calling :func:`_start_queue_drainer`.
 
+    Every payload passes here on its way to the web app, so this is where
+    oversized strings are trimmed, loosely, before they are queued
+    (:func:`~data.mesh_ingestor.field_limits.bound_post_payload`, SPEC SL8).
+    The caller's payload is never modified.
+
     .. note::
         The background drainer is used **only** when no custom ``send``
         override is provided (i.e. the production ``_post_json`` path).
@@ -499,6 +504,8 @@ def _queue_post_json(
     if send is None:
         send = _post_json
 
+    payload = field_limits.bound_post_payload(path, payload)
+
     if config.DEBUG:
         formatted_payload = (
             _payload_key_value_pairs(payload)