web: do not merge channels by ID in frontend (#637)

* web: do not merge channels by ID in frontend

* web: address review comments

* web: address review comments

README.md

@@ -252,15 +252,36 @@ services.potato-mesh = {
 
 ## Docker
 
-Docker images are published on Github for each release:
+Docker images are published on GitHub Container Registry for each release.
+Image names and tags follow the workflow format:
+`${IMAGE_PREFIX}-${service}-${architecture}:${tag}` (see `.github/workflows/docker.yml`).
 
 ```bash
-docker pull ghcr.io/l5yth/potato-mesh/web:latest      # newest release
-docker pull ghcr.io/l5yth/potato-mesh/web:v0.5.5      # pinned historical release
-docker pull ghcr.io/l5yth/potato-mesh/ingestor:latest
-docker pull ghcr.io/l5yth/potato-mesh/matrix-bridge:latest
+docker pull ghcr.io/l5yth/potato-mesh-web-linux-amd64:latest
+docker pull ghcr.io/l5yth/potato-mesh-web-linux-arm64:latest
+docker pull ghcr.io/l5yth/potato-mesh-web-linux-armv7:latest
+
+docker pull ghcr.io/l5yth/potato-mesh-ingestor-linux-amd64:latest
+docker pull ghcr.io/l5yth/potato-mesh-ingestor-linux-arm64:latest
+docker pull ghcr.io/l5yth/potato-mesh-ingestor-linux-armv7:latest
+
+docker pull ghcr.io/l5yth/potato-mesh-matrix-bridge-linux-amd64:latest
+docker pull ghcr.io/l5yth/potato-mesh-matrix-bridge-linux-arm64:latest
+docker pull ghcr.io/l5yth/potato-mesh-matrix-bridge-linux-armv7:latest
+
+# version-pinned examples
+docker pull ghcr.io/l5yth/potato-mesh-web-linux-amd64:v0.5.5
+docker pull ghcr.io/l5yth/potato-mesh-ingestor-linux-amd64:v0.5.5
+docker pull ghcr.io/l5yth/potato-mesh-matrix-bridge-linux-amd64:v0.5.5
 ```
 
+Note: `latest` is only published for non-prerelease versions. Pre-release tags
+such as `-rc`, `-beta`, `-alpha`, or `-dev` are version-tagged only.
+
+When using Compose, set `POTATOMESH_IMAGE_ARCH` in `docker-compose.yml` (or via
+environment) so service images resolve to the correct architecture variant and
+you avoid manual tag mistakes.
+
 Feel free to run the [configure.sh](./configure.sh) script to set up your
 environment. See the [Docker guide](DOCKER.md) for more details and custom
 deployment instructions.

data/mesh_ingestor/decode_payload.py

@@ -29,7 +29,6 @@ if SCRIPT_DIR in sys.path:
 from google.protobuf.json_format import MessageToDict
 from meshtastic.protobuf import mesh_pb2, telemetry_pb2
 
-
 PORTNUM_MAP: Dict[int, Tuple[str, Any]] = {
     3: ("POSITION_APP", mesh_pb2.Position),
     4: ("NODEINFO_APP", mesh_pb2.NodeInfo),

data/mesh_ingestor/handlers.py

@@ -424,6 +424,7 @@ def store_position_packet(packet: Mapping, decoded: Mapping) -> None:
         "hop_limit": hop_limit,
         "bitfield": bitfield,
         "payload_b64": payload_b64,
+        "ingestor": host_node_id(),
     }
     if raw_payload:
         position_payload["raw"] = raw_payload
@@ -568,6 +569,7 @@ def store_traceroute_packet(packet: Mapping, decoded: Mapping) -> None:
         "rssi": rssi,
         "snr": snr,
         "elapsed_ms": elapsed_ms,
+        "ingestor": host_node_id(),
     }
 
     _queue_post_json(
@@ -935,6 +937,7 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
         "rssi": rssi,
         "hop_limit": hop_limit,
         "payload_b64": payload_b64,
+        "ingestor": host_node_id(),
     }
 
     if battery_level is not None:
@@ -1263,6 +1266,7 @@ def store_neighborinfo_packet(packet: Mapping, decoded: Mapping) -> None:
         "neighbors": neighbor_entries,
         "rx_time": rx_time,
         "rx_iso": _iso(rx_time),
+        "ingestor": host_node_id(),
     }
 
     if node_broadcast_interval is not None:
@@ -1520,6 +1524,7 @@ def store_packet_dict(packet: Mapping) -> None:
         "hop_limit": int(hop) if hop is not None else None,
         "reply_id": reply_id,
         "emoji": emoji,
+        "ingestor": host_node_id(),
     }
 
     if not encrypted_flag and channel_name_value:

data/messages.sql

@@ -29,7 +29,8 @@ CREATE TABLE IF NOT EXISTS messages (
     modem_preset TEXT,
     channel_name TEXT,
     reply_id INTEGER,
-    emoji TEXT
+    emoji TEXT,
+    ingestor TEXT
 );
 
 CREATE INDEX IF NOT EXISTS idx_messages_rx_time   ON messages(rx_time);

data/neighbors.sql

@@ -17,6 +17,7 @@ CREATE TABLE IF NOT EXISTS neighbors (
     neighbor_id TEXT NOT NULL,
     snr         REAL,
     rx_time     INTEGER NOT NULL,
+    ingestor    TEXT,
     PRIMARY KEY (node_id, neighbor_id),
     FOREIGN KEY (node_id) REFERENCES nodes(node_id) ON DELETE CASCADE,
     FOREIGN KEY (neighbor_id) REFERENCES nodes(node_id) ON DELETE CASCADE

data/positions.sql

@@ -33,7 +33,8 @@ CREATE TABLE IF NOT EXISTS positions (
     rssi           INTEGER,
     hop_limit      INTEGER,
     bitfield       INTEGER,
-    payload_b64    TEXT
+    payload_b64    TEXT,
+    ingestor       TEXT
 );
 
 CREATE INDEX IF NOT EXISTS idx_positions_rx_time ON positions(rx_time);

data/telemetry.sql

@@ -53,7 +53,8 @@ CREATE TABLE IF NOT EXISTS telemetry (
     rainfall_1h             REAL,
     rainfall_24h            REAL,
     soil_moisture           INTEGER,
-    soil_temperature        REAL
+    soil_temperature        REAL,
+    ingestor                TEXT
 );
 
 CREATE INDEX IF NOT EXISTS idx_telemetry_rx_time ON telemetry(rx_time);

data/traces.sql

@@ -21,7 +21,8 @@ CREATE TABLE IF NOT EXISTS traces (
     rx_iso      TEXT NOT NULL,
     rssi        INTEGER,
     snr         REAL,
-    elapsed_ms  INTEGER
+    elapsed_ms  INTEGER,
+    ingestor    TEXT
 );
 
 CREATE TABLE IF NOT EXISTS trace_hops (

docker-compose.yml

@@ -81,7 +81,12 @@ x-matrix-bridge-base: &matrix-bridge-base
   image: ghcr.io/l5yth/potato-mesh-matrix-bridge-${POTATOMESH_IMAGE_ARCH:-linux-amd64}:${POTATOMESH_IMAGE_TAG:-latest}
   volumes:
     - potatomesh_matrix_bridge_state:/app
-    - ./matrix/Config.toml:/app/Config.toml:ro
+    - type: bind
+      source: ./matrix/Config.toml
+      target: /app/Config.toml
+      read_only: true
+      bind:
+        create_host_path: false
   restart: unless-stopped
   deploy:
     resources:
@@ -128,6 +133,8 @@ services:
   matrix-bridge:
     <<: *matrix-bridge-base
     network_mode: host
+    profiles:
+      - matrix
     depends_on:
       - web
     extra_hosts:

matrix/Cargo.lock

@@ -169,9 +169,9 @@ checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510"
 
 [[package]]
 name = "bytes"
-version = "1.11.0"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3"
+checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
 
 [[package]]
 name = "cc"

matrix/README.md

@@ -146,6 +146,38 @@ Container detection checks `POTATOMESH_CONTAINER`, `CONTAINER`, and `/proc/1/cgr
 
 Set `POTATOMESH_CONTAINER=0` or `--no-container` to opt out of container defaults.
 
+### Docker Compose First Run
+
+Before starting Compose, complete this preflight checklist:
+
+1. Ensure `matrix/Config.toml` exists as a regular file on the host (not a directory).
+2. Fill required Matrix values in `matrix/Config.toml`:
+   - `matrix.as_token`
+   - `matrix.hs_token`
+   - `matrix.server_name`
+   - `matrix.room_id`
+   - `matrix.homeserver`
+
+This is required because the shared Compose anchor `x-matrix-bridge-base` mounts `./matrix/Config.toml` to `/app/Config.toml`.
+Then follow the token and namespace requirements in [Matrix Appservice Setup (Synapse example)](#matrix-appservice-setup-synapse-example).
+
+#### Troubleshooting
+
+| Symptom | Likely cause | What to check |
+| --- | --- | --- |
+| `Is a directory (os error 21)` | Host mount source became a directory | `matrix/Config.toml` was missing at mount time and got created as a directory on host. |
+| `M_UNKNOWN_TOKEN` / `401 Unauthorized` | Matrix appservice token mismatch | Verify `matrix.as_token` matches your appservice registration and setup in [Matrix Appservice Setup (Synapse example)](#matrix-appservice-setup-synapse-example). |
+
+#### Recovery from accidental `Config.toml` directory creation
+
+```bash
+# from repo root
+rm -rf matrix/Config.toml
+touch matrix/Config.toml
+# then edit matrix/Config.toml and set valid matrix.as_token, matrix.hs_token,
+# matrix.server_name, matrix.room_id, and matrix.homeserver before starting compose
+```
+
 ### PotatoMesh API
 
 The bridge assumes:

tests/test_mesh.py

@@ -788,6 +788,7 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch):
 
     mesh.config.LORA_FREQ = 868
     mesh.config.MODEM_PRESET = "MediumFast"
+    mesh.register_host_node_id("!f00dbabe")
 
     packet = {
         "id": 123,
@@ -823,6 +824,7 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch):
     assert payload["rssi"] == -70
     assert payload["reply_id"] is None
     assert payload["emoji"] is None
+    assert payload["ingestor"] == "!f00dbabe"
     assert payload["lora_freq"] == 868
     assert payload["modem_preset"] == "MediumFast"
     assert priority == mesh._MESSAGE_POST_PRIORITY
@@ -879,6 +881,7 @@ def test_store_packet_dict_posts_position(mesh_module, monkeypatch):
 
     mesh.config.LORA_FREQ = 868
     mesh.config.MODEM_PRESET = "MediumFast"
+    mesh.register_host_node_id("!f00dbabe")
 
     packet = {
         "id": 200498337,
@@ -946,6 +949,7 @@ def test_store_packet_dict_posts_position(mesh_module, monkeypatch):
     )
     assert payload["lora_freq"] == 868
     assert payload["modem_preset"] == "MediumFast"
+    assert payload["ingestor"] == "!f00dbabe"
     assert payload["raw"]["time"] == 1_758_624_189
 
 
@@ -960,6 +964,7 @@ def test_store_packet_dict_posts_neighborinfo(mesh_module, monkeypatch):
 
     mesh.config.LORA_FREQ = 868
     mesh.config.MODEM_PRESET = "MediumFast"
+    mesh.register_host_node_id("!f00dbabe")
 
     packet = {
         "id": 2049886869,
@@ -1004,6 +1009,7 @@ def test_store_packet_dict_posts_neighborinfo(mesh_module, monkeypatch):
     assert neighbors[2]["neighbor_num"] == 0x0BAD_C0DE
     assert payload["lora_freq"] == 868
     assert payload["modem_preset"] == "MediumFast"
+    assert payload["ingestor"] == "!f00dbabe"
 
 
 def test_store_packet_dict_handles_nodeinfo_packet(mesh_module, monkeypatch):
@@ -2282,6 +2288,7 @@ def test_store_packet_dict_handles_telemetry_packet(mesh_module, monkeypatch):
 
     mesh.config.LORA_FREQ = 868
     mesh.config.MODEM_PRESET = "MediumFast"
+    mesh.register_host_node_id("!f00dbabe")
 
     packet = {
         "id": 1_256_091_342,
@@ -2334,6 +2341,7 @@ def test_store_packet_dict_handles_telemetry_packet(mesh_module, monkeypatch):
     assert payload["current"] == pytest.approx(0.0715)
     assert payload["lora_freq"] == 868
     assert payload["modem_preset"] == "MediumFast"
+    assert payload["ingestor"] == "!f00dbabe"
 
 
 def test_store_packet_dict_handles_environment_telemetry(mesh_module, monkeypatch):
@@ -2477,6 +2485,7 @@ def test_store_packet_dict_handles_traceroute_packet(mesh_module, monkeypatch):
 
     mesh.config.LORA_FREQ = 915
     mesh.config.MODEM_PRESET = "LongFast"
+    mesh.register_host_node_id("!f00dbabe")
 
     packet = {
         "id": 2_934_054_466,
@@ -2518,6 +2527,7 @@ def test_store_packet_dict_handles_traceroute_packet(mesh_module, monkeypatch):
     assert "elapsed_ms" in payload
     assert payload["lora_freq"] == 915
     assert payload["modem_preset"] == "LongFast"
+    assert payload["ingestor"] == "!f00dbabe"
 
 
 def test_traceroute_hop_normalization_supports_mappings(mesh_module, monkeypatch):

web/lib/potato_mesh/application.rb

@@ -139,7 +139,10 @@ module PotatoMesh
       set :public_folder, File.expand_path("../../public", __dir__)
       set :views, File.expand_path("../../views", __dir__)
       set :federation_thread, nil
+      set :initial_federation_thread, nil
       set :federation_worker_pool, nil
+      set :federation_shutdown_requested, false
+      set :federation_shutdown_hook_installed, false
       set :port, resolve_port
       set :bind, DEFAULT_BIND_ADDRESS
 

web/lib/potato_mesh/application/data_processing.rb

@@ -616,6 +616,7 @@ module PotatoMesh
 
         payload_b64 = string_or_nil(payload["payload_b64"] || payload["payload"])
         payload_b64 ||= string_or_nil(position_section.dig("payload", "__bytes_b64__"))
+        ingestor = string_or_nil(payload["ingestor"])
 
         row = [
           pos_id,
@@ -639,13 +640,14 @@ module PotatoMesh
           hop_limit,
           bitfield,
           payload_b64,
+          ingestor,
         ]
 
         with_busy_retry do
           db.execute <<~SQL, row
                        INSERT INTO positions(id,node_id,node_num,rx_time,rx_iso,position_time,to_id,latitude,longitude,altitude,location_source,
-                                             precision_bits,sats_in_view,pdop,ground_speed,ground_track,snr,rssi,hop_limit,bitfield,payload_b64)
-                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+                                             precision_bits,sats_in_view,pdop,ground_speed,ground_track,snr,rssi,hop_limit,bitfield,payload_b64,ingestor)
+                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                        ON CONFLICT(id) DO UPDATE SET
                          node_id=COALESCE(excluded.node_id,positions.node_id),
                          node_num=COALESCE(excluded.node_num,positions.node_num),
@@ -666,7 +668,8 @@ module PotatoMesh
                          rssi=COALESCE(excluded.rssi,positions.rssi),
                          hop_limit=COALESCE(excluded.hop_limit,positions.hop_limit),
                          bitfield=COALESCE(excluded.bitfield,positions.bitfield),
-                         payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64)
+                         payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64),
+                         ingestor=COALESCE(NULLIF(positions.ingestor,''), excluded.ingestor)
                      SQL
         end
 
@@ -721,6 +724,7 @@ module PotatoMesh
         touch_node_last_seen(db, node_id || node_num, node_num, rx_time: rx_time, source: :neighborinfo)
 
         neighbor_entries = []
+        ingestor = string_or_nil(payload["ingestor"])
         neighbors_payload = payload["neighbors"]
         neighbors_list = neighbors_payload.is_a?(Array) ? neighbors_payload : []
 
@@ -757,21 +761,41 @@ module PotatoMesh
           snr = coerce_float(neighbor["snr"])
 
           ensure_unknown_node(db, neighbor_id || neighbor_num, neighbor_num, heard_time: entry_rx_time)
-          touch_node_last_seen(db, neighbor_id || neighbor_num, neighbor_num, rx_time: entry_rx_time, source: :neighborinfo)
 
-          neighbor_entries << [neighbor_id, snr, entry_rx_time]
+          neighbor_entries << [neighbor_id, snr, entry_rx_time, ingestor]
         end
 
         with_busy_retry do
           db.transaction do
-            db.execute("DELETE FROM neighbors WHERE node_id = ?", [node_id])
-            neighbor_entries.each do |neighbor_id, snr_value, heard_time|
+            if neighbor_entries.empty?
+              db.execute("DELETE FROM neighbors WHERE node_id = ?", [node_id])
+            else
+              expected_neighbors = neighbor_entries.map(&:first).uniq
+              existing_neighbors = db.execute(
+                "SELECT neighbor_id FROM neighbors WHERE node_id = ?",
+                [node_id],
+              ).flatten
+              stale_neighbors = existing_neighbors - expected_neighbors
+              stale_neighbors.each_slice(500) do |slice|
+                placeholders = slice.map { "?" }.join(",")
+                db.execute(
+                  "DELETE FROM neighbors WHERE node_id = ? AND neighbor_id IN (#{placeholders})",
+                  [node_id] + slice,
+                )
+              end
+            end
+
+            neighbor_entries.each do |neighbor_id, snr_value, heard_time, reporter_id|
               db.execute(
                 <<~SQL,
-                INSERT OR REPLACE INTO neighbors(node_id, neighbor_id, snr, rx_time)
-                VALUES (?, ?, ?, ?)
+                INSERT INTO neighbors(node_id, neighbor_id, snr, rx_time, ingestor)
+                VALUES (?, ?, ?, ?, ?)
+                ON CONFLICT(node_id, neighbor_id) DO UPDATE SET
+                  snr = excluded.snr,
+                  rx_time = excluded.rx_time,
+                  ingestor = COALESCE(NULLIF(neighbors.ingestor,''), excluded.ingestor)
               SQL
-                [node_id, neighbor_id, snr_value, heard_time],
+                [node_id, neighbor_id, snr_value, heard_time, reporter_id],
               )
             end
           end
@@ -981,6 +1005,7 @@ module PotatoMesh
         payload_b64 = string_or_nil(payload["payload_b64"] || payload["payload"])
         lora_freq = coerce_integer(payload["lora_freq"] || payload["loraFrequency"])
         modem_preset = string_or_nil(payload["modem_preset"] || payload["modemPreset"])
+        ingestor = string_or_nil(payload["ingestor"])
 
         telemetry_section = normalize_json_object(payload["telemetry"])
         device_metrics = normalize_json_object(payload["device_metrics"] || payload["deviceMetrics"])
@@ -1310,6 +1335,7 @@ module PotatoMesh
           rainfall_24h,
           soil_moisture,
           soil_temperature,
+          ingestor,
         ]
 
         placeholders = Array.new(row.length, "?").join(",")
@@ -1317,7 +1343,7 @@ module PotatoMesh
         with_busy_retry do
           db.execute <<~SQL, row
                        INSERT INTO telemetry(id,node_id,node_num,from_id,to_id,rx_time,rx_iso,telemetry_time,channel,portnum,hop_limit,snr,rssi,bitfield,payload_b64,
-                                             battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds,temperature,relative_humidity,barometric_pressure,gas_resistance,current,iaq,distance,lux,white_lux,ir_lux,uv_lux,wind_direction,wind_speed,weight,wind_gust,wind_lull,radiation,rainfall_1h,rainfall_24h,soil_moisture,soil_temperature)
+                                             battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds,temperature,relative_humidity,barometric_pressure,gas_resistance,current,iaq,distance,lux,white_lux,ir_lux,uv_lux,wind_direction,wind_speed,weight,wind_gust,wind_lull,radiation,rainfall_1h,rainfall_24h,soil_moisture,soil_temperature,ingestor)
                        VALUES (#{placeholders})
                        ON CONFLICT(id) DO UPDATE SET
                          node_id=COALESCE(excluded.node_id,telemetry.node_id),
@@ -1359,7 +1385,8 @@ module PotatoMesh
                          rainfall_1h=COALESCE(excluded.rainfall_1h,telemetry.rainfall_1h),
                          rainfall_24h=COALESCE(excluded.rainfall_24h,telemetry.rainfall_24h),
                          soil_moisture=COALESCE(excluded.soil_moisture,telemetry.soil_moisture),
-                         soil_temperature=COALESCE(excluded.soil_temperature,telemetry.soil_temperature)
+                         soil_temperature=COALESCE(excluded.soil_temperature,telemetry.soil_temperature),
+                         ingestor=COALESCE(NULLIF(telemetry.ingestor,''), excluded.ingestor)
                      SQL
         end
 
@@ -1410,6 +1437,7 @@ module PotatoMesh
             metrics&.[]("latency_ms") ||
             metrics&.[]("latencyMs"),
         )
+        ingestor = string_or_nil(payload["ingestor"])
 
         hops_value = payload.key?("hops") ? payload["hops"] : payload["path"]
         hops = normalize_trace_hops(hops_value)
@@ -1421,9 +1449,9 @@ module PotatoMesh
         end
 
         with_busy_retry do
-          db.execute <<~SQL, [trace_identifier, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms]
-                       INSERT INTO traces(id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms)
-                            VALUES(?,?,?,?,?,?,?,?,?)
+          db.execute <<~SQL, [trace_identifier, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, ingestor]
+                       INSERT INTO traces(id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, ingestor)
+                            VALUES(?,?,?,?,?,?,?,?,?,?)
                        ON CONFLICT(id) DO UPDATE SET
                          request_id=COALESCE(excluded.request_id,traces.request_id),
                          src=COALESCE(excluded.src,traces.src),
@@ -1432,7 +1460,8 @@ module PotatoMesh
                          rx_iso=excluded.rx_iso,
                          rssi=COALESCE(excluded.rssi,traces.rssi),
                          snr=COALESCE(excluded.snr,traces.snr),
-                         elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms)
+                         elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms),
+                         ingestor=COALESCE(NULLIF(traces.ingestor,''), excluded.ingestor)
                      SQL
 
           trace_id = trace_identifier || db.last_insert_row_id
@@ -1566,7 +1595,6 @@ module PotatoMesh
         channel_index = coerce_integer(message["channel"] || message["channel_index"] || message["channelIndex"])
 
         decrypted_payload = nil
-        decrypted_text = nil
         decrypted_portnum = nil
 
         if encrypted && (text.nil? || text.to_s.strip.empty?)
@@ -1581,19 +1609,6 @@ module PotatoMesh
           if decrypted
             decrypted_payload = decrypted
             decrypted_portnum = decrypted[:portnum]
-
-            if decrypted[:text]
-              text = decrypted[:text]
-              decrypted_text = text
-              clear_encrypted = true
-              encrypted = nil
-              message["text"] = text
-              message["channel_name"] ||= decrypted[:channel_name]
-              if portnum.nil? && decrypted_portnum
-                portnum = decrypted_portnum
-                message["portnum"] = portnum
-              end
-            end
           end
         end
 
@@ -1607,6 +1622,7 @@ module PotatoMesh
         channel_name = string_or_nil(message["channel_name"] || message["channelName"])
         reply_id = coerce_integer(message["reply_id"] || message["replyId"])
         emoji = string_or_nil(message["emoji"])
+        ingestor = string_or_nil(message["ingestor"])
 
         row = [
           msg_id,
@@ -1626,11 +1642,12 @@ module PotatoMesh
           channel_name,
           reply_id,
           emoji,
+          ingestor,
         ]
 
         with_busy_retry do
           existing = db.get_first_row(
-            "SELECT from_id, to_id, text, encrypted, lora_freq, modem_preset, channel_name, reply_id, emoji, portnum FROM messages WHERE id = ?",
+            "SELECT from_id, to_id, text, encrypted, lora_freq, modem_preset, channel_name, reply_id, emoji, portnum, ingestor FROM messages WHERE id = ?",
             [msg_id],
           )
           if existing
@@ -1728,6 +1745,12 @@ module PotatoMesh
               updates["emoji"] = emoji if should_update
             end
 
+            if ingestor
+              existing_ingestor = existing.is_a?(Hash) ? existing["ingestor"] : existing[10]
+              existing_ingestor = string_or_nil(existing_ingestor)
+              updates["ingestor"] = ingestor if existing_ingestor.nil?
+            end
+
             unless updates.empty?
               assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ")
               db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [msg_id])
@@ -1737,12 +1760,12 @@ module PotatoMesh
 
             begin
               db.execute <<~SQL, row
-                           INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,portnum,text,encrypted,snr,rssi,hop_limit,lora_freq,modem_preset,channel_name,reply_id,emoji)
-                           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+                           INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,portnum,text,encrypted,snr,rssi,hop_limit,lora_freq,modem_preset,channel_name,reply_id,emoji,ingestor)
+                           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                          SQL
             rescue SQLite3::ConstraintException
               existing_row = db.get_first_row(
-                "SELECT text, encrypted FROM messages WHERE id = ?",
+                "SELECT text, encrypted, ingestor FROM messages WHERE id = ?",
                 [msg_id],
               )
               existing_text = existing_row.is_a?(Hash) ? existing_row["text"] : existing_row&.[](0)
@@ -1750,6 +1773,8 @@ module PotatoMesh
               allow_encrypted_update = existing_text_str.nil? || existing_text_str.strip.empty?
               existing_encrypted = existing_row.is_a?(Hash) ? existing_row["encrypted"] : existing_row&.[](1)
               existing_encrypted_str = existing_encrypted&.to_s
+              existing_ingestor = existing_row.is_a?(Hash) ? existing_row["ingestor"] : existing_row&.[](2)
+              existing_ingestor = string_or_nil(existing_ingestor)
               decrypted_precedence = text && (clear_encrypted || (existing_encrypted_str && !existing_encrypted_str.strip.empty?))
 
               fallback_updates = {}
@@ -1777,6 +1802,7 @@ module PotatoMesh
               end
               fallback_updates["reply_id"] = reply_id unless reply_id.nil?
               fallback_updates["emoji"] = emoji if emoji
+              fallback_updates["ingestor"] = ingestor if ingestor && existing_ingestor.nil?
               unless fallback_updates.empty?
                 assignments = fallback_updates.keys.map { |column| "#{column} = ?" }.join(", ")
                 db.execute("UPDATE messages SET #{assignments} WHERE id = ?", fallback_updates.values + [msg_id])
@@ -1785,7 +1811,7 @@ module PotatoMesh
           end
         end
 
-        if clear_encrypted && decrypted_text
+        if clear_encrypted && text
           debug_log(
             "Stored decrypted text message",
             context: "data_processing.insert_message",
@@ -1827,7 +1853,7 @@ module PotatoMesh
           )
         end
 
-        should_touch_message = !stored_decrypted || decrypted_text
+        should_touch_message = !stored_decrypted
         if should_touch_message
           ensure_unknown_node(db, from_id || raw_from_id, message["from_num"], heard_time: rx_time)
           touch_node_last_seen(
@@ -1893,7 +1919,7 @@ module PotatoMesh
         return false unless portnum_value
 
         payload_b64 = Base64.strict_encode64(payload_bytes)
-        supported_ports = [3, 67, 70, 71]
+        supported_ports = [3, 4, 67, 70, 71]
         return false unless supported_ports.include?(portnum_value)
 
         decoded = PotatoMesh::App::Meshtastic::PayloadDecoder.decode(
@@ -1918,6 +1944,7 @@ module PotatoMesh
           "lora_freq" => coerce_integer(message["lora_freq"] || message["loraFrequency"]),
           "modem_preset" => string_or_nil(message["modem_preset"] || message["modemPreset"]),
           "payload_b64" => payload_b64,
+          "ingestor" => string_or_nil(message["ingestor"]),
         }
 
         case decoded["type"]
@@ -1931,6 +1958,33 @@ module PotatoMesh
             portnum: portnum_value,
           )
           true
+        when "NODEINFO_APP"
+          node_payload = normalize_decrypted_nodeinfo_payload(decoded["payload"])
+          return false unless valid_decrypted_nodeinfo_payload?(node_payload)
+
+          node_id = string_or_nil(node_payload["id"]) || from_id
+          node_num = coerce_integer(node_payload["num"]) ||
+                     coerce_integer(message["from_num"]) ||
+                     resolve_node_num(from_id, message)
+          node_id ||= format("!%08x", node_num & 0xFFFFFFFF) if node_num
+          return false unless node_id
+
+          payload = node_payload.merge(
+            "num" => node_num,
+            "lastHeard" => coerce_integer(node_payload["lastHeard"] || node_payload["last_heard"]) || rx_time,
+            "snr" => node_payload.key?("snr") ? node_payload["snr"] : snr,
+            "lora_freq" => common_payload["lora_freq"],
+            "modem_preset" => common_payload["modem_preset"],
+          )
+          upsert_node(db, node_id, payload)
+          debug_log(
+            "Stored decrypted node payload",
+            context: "data_processing.store_decrypted_payload",
+            message_id: packet_id,
+            portnum: portnum_value,
+            node_id: node_id,
+          )
+          true
         when "TELEMETRY_APP"
           payload = common_payload.merge("telemetry" => decoded["payload"])
           insert_telemetry(db, payload)
@@ -1993,6 +2047,92 @@ module PotatoMesh
         end
       end
 
+      # Validate decoded NodeInfo payloads before upserting node records.
+      #
+      # @param payload [Object] decoded payload candidate.
+      # @return [Boolean] true when the payload resembles a Meshtastic NodeInfo.
+      def valid_decrypted_nodeinfo_payload?(payload)
+        return false unless payload.is_a?(Hash)
+        return false if payload.empty?
+        return false unless payload["user"].is_a?(Hash)
+
+        return false if payload.key?("position") && !payload["position"].is_a?(Hash)
+        return false if payload.key?("deviceMetrics") && !payload["deviceMetrics"].is_a?(Hash)
+        return false unless nodeinfo_user_has_identifying_fields?(payload["user"])
+
+        true
+      end
+
+      # Normalize decoded NodeInfo payload keys for +upsert_node+ compatibility.
+      #
+      # The Python decoder preserves protobuf field names, so nested hashes may
+      # use +snake_case+ keys that +upsert_node+ does not read.
+      #
+      # @param payload [Object] decoded NodeInfo payload.
+      # @return [Hash] normalized payload hash.
+      def normalize_decrypted_nodeinfo_payload(payload)
+        return {} unless payload.is_a?(Hash)
+
+        user = payload["user"]
+        normalized_user = user.is_a?(Hash) ? user.dup : nil
+        if normalized_user
+          normalized_user["shortName"] ||= normalized_user["short_name"]
+          normalized_user["longName"] ||= normalized_user["long_name"]
+          normalized_user["hwModel"] ||= normalized_user["hw_model"]
+          normalized_user["publicKey"] ||= normalized_user["public_key"]
+          normalized_user["isUnmessagable"] = normalized_user["is_unmessagable"] if normalized_user.key?("is_unmessagable")
+        end
+
+        metrics = payload["deviceMetrics"] || payload["device_metrics"]
+        normalized_metrics = metrics.is_a?(Hash) ? metrics.dup : nil
+        if normalized_metrics
+          normalized_metrics["batteryLevel"] ||= normalized_metrics["battery_level"]
+          normalized_metrics["channelUtilization"] ||= normalized_metrics["channel_utilization"]
+          normalized_metrics["airUtilTx"] ||= normalized_metrics["air_util_tx"]
+          normalized_metrics["uptimeSeconds"] ||= normalized_metrics["uptime_seconds"]
+        end
+
+        position = payload["position"]
+        normalized_position = position.is_a?(Hash) ? position.dup : nil
+        if normalized_position
+          normalized_position["precisionBits"] ||= normalized_position["precision_bits"]
+          normalized_position["locationSource"] ||= normalized_position["location_source"]
+        end
+
+        normalized = payload.dup
+        normalized["user"] = normalized_user if normalized_user
+        normalized["deviceMetrics"] = normalized_metrics if normalized_metrics
+        normalized["position"] = normalized_position if normalized_position
+        normalized["lastHeard"] ||= normalized["last_heard"]
+        normalized["hopsAway"] ||= normalized["hops_away"]
+        normalized["isFavorite"] = normalized["is_favorite"] if normalized.key?("is_favorite")
+        normalized["hwModel"] ||= normalized["hw_model"]
+        normalized
+      end
+
+      # Validate that a decoded NodeInfo user section contains identifying data.
+      #
+      # @param user [Hash] decoded NodeInfo user payload.
+      # @return [Boolean] true when at least one identifying field is present.
+      def nodeinfo_user_has_identifying_fields?(user)
+        identifying_fields = [
+          user["id"],
+          user["shortName"],
+          user["short_name"],
+          user["longName"],
+          user["long_name"],
+          user["macaddr"],
+          user["hwModel"],
+          user["hw_model"],
+          user["publicKey"],
+          user["public_key"],
+        ]
+
+        identifying_fields.any? do |value|
+          value.is_a?(String) ? !value.strip.empty? : !value.nil?
+        end
+      end
+
       def normalize_node_id(db, node_ref)
         return nil if node_ref.nil?
         ref_str = node_ref.to_s.strip

web/lib/potato_mesh/application/database.rb

@@ -149,6 +149,9 @@ module PotatoMesh
           db.execute("ALTER TABLE messages ADD COLUMN emoji TEXT")
           message_columns << "emoji"
         end
+        unless message_columns.include?("ingestor")
+          db.execute("ALTER TABLE messages ADD COLUMN ingestor TEXT")
+        end
 
         reply_index_exists =
           db.get_first_value(
@@ -188,6 +191,31 @@ module PotatoMesh
           db.execute("ALTER TABLE telemetry ADD COLUMN #{name} #{type}")
           telemetry_columns << name
         end
+        unless telemetry_columns.include?("ingestor")
+          db.execute("ALTER TABLE telemetry ADD COLUMN ingestor TEXT")
+        end
+
+        position_tables =
+          db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='positions'").flatten
+        if position_tables.empty?
+          positions_schema = File.expand_path("../../../../data/positions.sql", __dir__)
+          db.execute_batch(File.read(positions_schema))
+        end
+        position_columns = db.execute("PRAGMA table_info(positions)").map { |row| row[1] }
+        unless position_columns.include?("ingestor")
+          db.execute("ALTER TABLE positions ADD COLUMN ingestor TEXT")
+        end
+
+        neighbor_tables =
+          db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='neighbors'").flatten
+        if neighbor_tables.empty?
+          neighbors_schema = File.expand_path("../../../../data/neighbors.sql", __dir__)
+          db.execute_batch(File.read(neighbors_schema))
+        end
+        neighbor_columns = db.execute("PRAGMA table_info(neighbors)").map { |row| row[1] }
+        unless neighbor_columns.include?("ingestor")
+          db.execute("ALTER TABLE neighbors ADD COLUMN ingestor TEXT")
+        end
 
         trace_tables =
           db.execute(
@@ -197,6 +225,10 @@ module PotatoMesh
           traces_schema = File.expand_path("../../../../data/traces.sql", __dir__)
           db.execute_batch(File.read(traces_schema))
         end
+        trace_columns = db.execute("PRAGMA table_info(traces)").map { |row| row[1] }
+        unless trace_columns.include?("ingestor")
+          db.execute("ALTER TABLE traces ADD COLUMN ingestor TEXT")
+        end
 
         ingestor_tables =
           db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestors'").flatten

web/lib/potato_mesh/application/federation.rb

@@ -17,6 +17,8 @@
 module PotatoMesh
   module App
     module Federation
+      FEDERATION_SLEEP_SLICE_SECONDS = 0.2
+
       # Resolve the canonical domain for the running instance.
       #
       # @return [String, nil] sanitized instance domain or nil outside production.
@@ -170,6 +172,9 @@ module PotatoMesh
       # @return [PotatoMesh::App::WorkerPool, nil] active worker pool if created.
       def ensure_federation_worker_pool!
         return nil unless federation_enabled?
+        return nil if federation_shutdown_requested?
+
+        ensure_federation_shutdown_hook!
 
         existing = settings.respond_to?(:federation_worker_pool) ? settings.federation_worker_pool : nil
         return existing if existing&.alive?
@@ -181,16 +186,77 @@ module PotatoMesh
           name: "potato-mesh-fed",
         )
 
+        set(:federation_worker_pool, pool) if respond_to?(:set)
+        pool
+      end
+
+      # Ensure federation background workers are torn down during process exit.
+      #
+      # @return [void]
+      def ensure_federation_shutdown_hook!
+        application = is_a?(Class) ? self : self.class
+        return application.ensure_federation_shutdown_hook! unless application.equal?(self)
+
+        installed = if respond_to?(:settings) && settings.respond_to?(:federation_shutdown_hook_installed)
+            settings.federation_shutdown_hook_installed
+          else
+            instance_variable_defined?(:@federation_shutdown_hook_installed) && @federation_shutdown_hook_installed
+          end
+        return if installed
+
+        if respond_to?(:set) && settings.respond_to?(:federation_shutdown_hook_installed=)
+          set(:federation_shutdown_hook_installed, true)
+        else
+          @federation_shutdown_hook_installed = true
+        end
+
         at_exit do
           begin
-            pool.shutdown(timeout: PotatoMesh::Config.federation_task_timeout_seconds)
+            application.shutdown_federation_background_work!(timeout: PotatoMesh::Config.federation_shutdown_timeout_seconds)
           rescue StandardError
             # Suppress shutdown errors during interpreter teardown.
           end
         end
+      end
 
-        set(:federation_worker_pool, pool) if respond_to?(:set)
-        pool
+      # Check whether federation workers have received a shutdown request.
+      #
+      # @return [Boolean] true when stop has been requested.
+      def federation_shutdown_requested?
+        return false unless respond_to?(:settings)
+        return false unless settings.respond_to?(:federation_shutdown_requested)
+
+        settings.federation_shutdown_requested == true
+      end
+
+      # Mark federation background work as shutting down.
+      #
+      # @return [void]
+      def request_federation_shutdown!
+        set(:federation_shutdown_requested, true) if respond_to?(:set)
+      end
+
+      # Clear any previously requested federation shutdown marker.
+      #
+      # @return [void]
+      def clear_federation_shutdown_request!
+        set(:federation_shutdown_requested, false) if respond_to?(:set)
+      end
+
+      # Sleep in short intervals so federation loops can react to shutdown.
+      #
+      # @param seconds [Numeric] target sleep duration.
+      # @return [Boolean] true when the full delay elapsed without shutdown.
+      def federation_sleep_with_shutdown(seconds)
+        remaining = seconds.to_f
+        while remaining.positive?
+          return false if federation_shutdown_requested?
+
+          slice = [remaining, FEDERATION_SLEEP_SLICE_SECONDS].min
+          Kernel.sleep(slice)
+          remaining -= slice
+        end
+        !federation_shutdown_requested?
       end
 
       # Shutdown and clear the federation worker pool if present.
@@ -214,6 +280,44 @@ module PotatoMesh
         end
       end
 
+      # Gracefully terminate federation background loops and worker pool tasks.
+      #
+      # @param timeout [Numeric, nil] maximum join time applied per thread.
+      # @return [void]
+      def shutdown_federation_background_work!(timeout: nil)
+        request_federation_shutdown!
+        timeout_value = timeout || PotatoMesh::Config.federation_shutdown_timeout_seconds
+        stop_federation_thread!(:initial_federation_thread, timeout: timeout_value)
+        stop_federation_thread!(:federation_thread, timeout: timeout_value)
+        shutdown_federation_worker_pool!
+        clear_federation_crawl_state!
+      end
+
+      # Stop a specific federation thread setting and clear its reference.
+      #
+      # @param setting_name [Symbol] settings key storing the thread object.
+      # @param timeout [Numeric] seconds to wait for clean thread exit.
+      # @return [void]
+      def stop_federation_thread!(setting_name, timeout:)
+        return unless respond_to?(:settings)
+        return unless settings.respond_to?(setting_name)
+
+        thread = settings.public_send(setting_name)
+        if thread&.alive?
+          begin
+            thread.wakeup if thread.respond_to?(:wakeup)
+          rescue ThreadError
+            # The thread may not currently be sleeping; continue shutdown.
+          end
+          thread.join(timeout)
+          if thread.alive?
+            thread.kill
+            thread.join(0.1)
+          end
+        end
+        set(setting_name, nil) if respond_to?(:set)
+      end
+
       def federation_target_domains(self_domain)
         normalized_self = sanitize_instance_domain(self_domain)&.downcase
         ordered = []
@@ -265,16 +369,21 @@ module PotatoMesh
 
       def announce_instance_to_domain(domain, payload_json)
         return false unless domain && !domain.empty?
+        return false if federation_shutdown_requested?
 
         https_failures = []
 
-        instance_uri_candidates(domain, "/api/instances").each do |uri|
+        published = instance_uri_candidates(domain, "/api/instances").any? do |uri|
+          break false if federation_shutdown_requested?
+
           begin
             http = build_remote_http_client(uri)
-            response = http.start do |connection|
-              request = build_federation_http_request(Net::HTTP::Post, uri)
-              request.body = payload_json
-              connection.request(request)
+            response = Timeout.timeout(PotatoMesh::Config.remote_instance_request_timeout) do
+              http.start do |connection|
+                request = build_federation_http_request(Net::HTTP::Post, uri)
+                request.body = payload_json
+                connection.request(request)
+              end
             end
             if response.is_a?(Net::HTTPSuccess)
               debug_log(
@@ -283,14 +392,16 @@ module PotatoMesh
                 target: uri.to_s,
                 status: response.code,
               )
-              return true
+              true
+            else
+              debug_log(
+                "Federation announcement failed",
+                context: "federation.announce",
+                target: uri.to_s,
+                status: response.code,
+              )
+              false
             end
-            debug_log(
-              "Federation announcement failed",
-              context: "federation.announce",
-              target: uri.to_s,
-              status: response.code,
-            )
           rescue StandardError => e
             metadata = {
               context: "federation.announce",
@@ -305,9 +416,18 @@ module PotatoMesh
                 **metadata,
               )
               https_failures << metadata
-              next
+            else
+              warn_log(
+                "Federation announcement raised exception",
+                **metadata,
+              )
             end
+            false
+          end
+        end
 
+        unless published
+          https_failures.each do |metadata|
             warn_log(
               "Federation announcement raised exception",
               **metadata,
@@ -315,14 +435,7 @@ module PotatoMesh
           end
         end
 
-        https_failures.each do |metadata|
-          warn_log(
-            "Federation announcement raised exception",
-            **metadata,
-          )
-        end
-
-        false
+        published
       end
 
       # Determine whether an HTTPS announcement failure should fall back to HTTP.
@@ -342,6 +455,7 @@ module PotatoMesh
 
       def announce_instance_to_all_domains
         return unless federation_enabled?
+        return if federation_shutdown_requested?
 
         attributes, signature = ensure_self_instance_record!
         payload_json = JSON.generate(instance_announcement_payload(attributes, signature))
@@ -349,13 +463,15 @@ module PotatoMesh
         pool = federation_worker_pool
         scheduled = []
 
-        domains.each do |domain|
+        domains.each_with_object(scheduled) do |domain, scheduled_tasks|
+          break if federation_shutdown_requested?
+
           if pool
             begin
               task = pool.schedule do
                 announce_instance_to_domain(domain, payload_json)
               end
-              scheduled << [domain, task]
+              scheduled_tasks << [domain, task]
               next
             rescue PotatoMesh::App::WorkerPool::QueueFullError
               warn_log(
@@ -396,7 +512,9 @@ module PotatoMesh
         return if scheduled.empty?
 
         timeout = PotatoMesh::Config.federation_task_timeout_seconds
-        scheduled.each do |domain, task|
+        scheduled.all? do |domain, task|
+          break false if federation_shutdown_requested?
+
           begin
             task.wait(timeout: timeout)
           rescue PotatoMesh::App::WorkerPool::TaskTimeoutError => e
@@ -417,19 +535,23 @@ module PotatoMesh
               error_message: e.message,
             )
           end
+          true
         end
       end
 
       def start_federation_announcer!
         # Federation broadcasts must not execute when federation support is disabled.
         return nil unless federation_enabled?
+        clear_federation_shutdown_request!
+        ensure_federation_shutdown_hook!
 
         existing = settings.federation_thread
         return existing if existing&.alive?
 
         thread = Thread.new do
           loop do
-            sleep PotatoMesh::Config.federation_announcement_interval
+            break unless federation_sleep_with_shutdown(PotatoMesh::Config.federation_announcement_interval)
+
             begin
               announce_instance_to_all_domains
             rescue StandardError => e
@@ -455,6 +577,8 @@ module PotatoMesh
       def start_initial_federation_announcement!
         # Skip the initial broadcast entirely when federation is disabled.
         return nil unless federation_enabled?
+        clear_federation_shutdown_request!
+        ensure_federation_shutdown_hook!
 
         existing = settings.respond_to?(:initial_federation_thread) ? settings.initial_federation_thread : nil
         return existing if existing&.alive?
@@ -462,7 +586,12 @@ module PotatoMesh
         thread = Thread.new do
           begin
             delay = PotatoMesh::Config.initial_federation_delay_seconds
-            Kernel.sleep(delay) if delay.positive?
+            if delay.positive?
+              completed = federation_sleep_with_shutdown(delay)
+              next unless completed
+            end
+            next if federation_shutdown_requested?
+
             announce_instance_to_all_domains
           rescue StandardError => e
             warn_log(
@@ -523,15 +652,19 @@ module PotatoMesh
       end
 
       def perform_instance_http_request(uri)
+        raise InstanceFetchError, "federation shutdown requested" if federation_shutdown_requested?
+
         http = build_remote_http_client(uri)
-        http.start do |connection|
-          request = build_federation_http_request(Net::HTTP::Get, uri)
-          response = connection.request(request)
-          case response
-          when Net::HTTPSuccess
-            response.body
-          else
-            raise InstanceFetchError, "unexpected response #{response.code}"
+        Timeout.timeout(PotatoMesh::Config.remote_instance_request_timeout) do
+          http.start do |connection|
+            request = build_federation_http_request(Net::HTTP::Get, uri)
+            response = connection.request(request)
+            case response
+            when Net::HTTPSuccess
+              response.body
+            else
+              raise InstanceFetchError, "unexpected response #{response.code}"
+            end
           end
         end
       rescue StandardError => e
@@ -588,8 +721,12 @@ module PotatoMesh
       end
 
       def fetch_instance_json(domain, path)
+        return [nil, ["federation shutdown requested"]] if federation_shutdown_requested?
+
         errors = []
         instance_uri_candidates(domain, path).each do |uri|
+          break if federation_shutdown_requested?
+
           begin
             body = perform_instance_http_request(uri)
             return [JSON.parse(body), uri] if body
@@ -662,51 +799,149 @@ module PotatoMesh
       # @param overall_limit [Integer, nil] maximum unique domains visited.
       # @return [Boolean] true when the crawl was scheduled successfully.
       def enqueue_federation_crawl(domain, per_response_limit:, overall_limit:)
-        pool = federation_worker_pool
+        sanitized_domain = sanitize_instance_domain(domain)
+        unless sanitized_domain
+          warn_log(
+            "Skipped remote instance crawl",
+            context: "federation.instances",
+            domain: domain,
+            reason: "invalid domain",
+          )
+          return false
+        end
+        return false if federation_shutdown_requested?
+
+        application = is_a?(Class) ? self : self.class
+        pool = application.federation_worker_pool
         unless pool
           debug_log(
             "Skipped remote instance crawl",
             context: "federation.instances",
-            domain: domain,
+            domain: sanitized_domain,
             reason: "federation disabled",
           )
           return false
         end
 
-        application = is_a?(Class) ? self : self.class
+        claim_result = application.claim_federation_crawl_slot(sanitized_domain)
+        unless claim_result == :claimed
+          debug_log(
+            "Skipped remote instance crawl",
+            context: "federation.instances",
+            domain: sanitized_domain,
+            reason: claim_result == :in_flight ? "crawl already in flight" : "recent crawl completed",
+          )
+          return false
+        end
+
         pool.schedule do
-          db = application.open_database
+          db = nil
           begin
+            db = application.open_database
             application.ingest_known_instances_from!(
               db,
-              domain,
+              sanitized_domain,
               per_response_limit: per_response_limit,
               overall_limit: overall_limit,
             )
           ensure
             db&.close
+            application.release_federation_crawl_slot(sanitized_domain)
           end
         end
 
         true
       rescue PotatoMesh::App::WorkerPool::QueueFullError
-        warn_log(
-          "Skipped remote instance crawl",
-          context: "federation.instances",
-          domain: domain,
-          reason: "worker queue saturated",
-        )
-        false
+        application.handle_failed_federation_crawl_schedule(sanitized_domain, "worker queue saturated")
       rescue PotatoMesh::App::WorkerPool::ShutdownError
+        application.handle_failed_federation_crawl_schedule(sanitized_domain, "worker pool shut down")
+      end
+
+      # Handle a failed crawl schedule attempt without applying cooldown.
+      #
+      # @param domain [String] canonical domain that failed to schedule.
+      # @param reason [String] human-readable failure reason.
+      # @return [Boolean] always false because scheduling did not succeed.
+      def handle_failed_federation_crawl_schedule(domain, reason)
+        release_federation_crawl_slot(domain, record_completion: false)
         warn_log(
           "Skipped remote instance crawl",
           context: "federation.instances",
           domain: domain,
-          reason: "worker pool shut down",
+          reason: reason,
         )
         false
       end
 
+      # Initialize shared in-memory state used to deduplicate crawl scheduling.
+      #
+      # @return [void]
+      def initialize_federation_crawl_state!
+        @federation_crawl_init_mutex ||= Mutex.new
+        return if instance_variable_defined?(:@federation_crawl_mutex) && @federation_crawl_mutex
+
+        @federation_crawl_init_mutex.synchronize do
+          return if instance_variable_defined?(:@federation_crawl_mutex) && @federation_crawl_mutex
+
+          @federation_crawl_mutex = Mutex.new
+          @federation_crawl_in_flight = Set.new
+          @federation_crawl_last_completed_at = {}
+        end
+      end
+
+      # Retrieve the cooldown period used for duplicate crawl suppression.
+      #
+      # @return [Integer] seconds a domain remains in cooldown after completion.
+      def federation_crawl_cooldown_seconds
+        PotatoMesh::Config.federation_crawl_cooldown_seconds
+      end
+
+      # Mark a domain crawl as claimed if no active or recent crawl exists.
+      #
+      # @param domain [String] canonical domain name.
+      # @return [Symbol] +:claimed+, +:in_flight+, or +:cooldown+.
+      def claim_federation_crawl_slot(domain)
+        initialize_federation_crawl_state!
+        now = Time.now.to_i
+        @federation_crawl_mutex.synchronize do
+          return :in_flight if @federation_crawl_in_flight.include?(domain)
+
+          last_completed = @federation_crawl_last_completed_at[domain]
+          if last_completed && now - last_completed < federation_crawl_cooldown_seconds
+            return :cooldown
+          end
+
+          @federation_crawl_in_flight << domain
+          :claimed
+        end
+      end
+
+      # Release an in-flight crawl claim and record completion timestamp.
+      #
+      # @param domain [String] canonical domain name.
+      # @param record_completion [Boolean] true to apply cooldown tracking.
+      # @return [void]
+      def release_federation_crawl_slot(domain, record_completion: true)
+        return unless domain
+
+        initialize_federation_crawl_state!
+        @federation_crawl_mutex.synchronize do
+          @federation_crawl_in_flight.delete(domain)
+          @federation_crawl_last_completed_at[domain] = Time.now.to_i if record_completion
+        end
+      end
+
+      # Clear all in-memory crawl scheduling state.
+      #
+      # @return [void]
+      def clear_federation_crawl_state!
+        initialize_federation_crawl_state!
+        @federation_crawl_mutex.synchronize do
+          @federation_crawl_in_flight.clear
+          @federation_crawl_last_completed_at.clear
+        end
+      end
+
       # Recursively ingest federation records exposed by the supplied domain.
       #
       # @param db [SQLite3::Database] open database connection used for writes.
@@ -724,6 +959,7 @@ module PotatoMesh
       )
         sanitized = sanitize_instance_domain(domain)
         return visited || Set.new unless sanitized
+        return visited || Set.new if federation_shutdown_requested?
 
         visited ||= Set.new
 
@@ -758,6 +994,8 @@ module PotatoMesh
         processed_entries = 0
         recent_cutoff = Time.now.to_i - PotatoMesh::Config.remote_instance_max_node_age
         payload.each do |entry|
+          break if federation_shutdown_requested?
+
           if per_response_limit && per_response_limit.positive? && processed_entries >= per_response_limit
             debug_log(
               "Skipped remote instance entry due to response limit",

web/lib/potato_mesh/application/queries.rb

@@ -357,7 +357,7 @@ module PotatoMesh
           SELECT m.id, m.rx_time, m.rx_iso, m.from_id, m.to_id, m.channel,
                  m.portnum, m.text, m.encrypted, m.rssi, m.hop_limit,
                  m.lora_freq, m.modem_preset, m.channel_name, m.snr,
-                 m.reply_id, m.emoji
+                 m.reply_id, m.emoji, m.ingestor
           FROM messages m
         SQL
         sql += "    WHERE #{where_clauses.join(" AND ")}\n"

web/lib/potato_mesh/config.rb

@@ -37,11 +37,14 @@ module PotatoMesh
     DEFAULT_MAX_DISTANCE_KM = 42.0
     DEFAULT_REMOTE_INSTANCE_CONNECT_TIMEOUT = 15
     DEFAULT_REMOTE_INSTANCE_READ_TIMEOUT = 60
+    DEFAULT_REMOTE_INSTANCE_REQUEST_TIMEOUT = 30
     DEFAULT_FEDERATION_MAX_INSTANCES_PER_RESPONSE = 64
     DEFAULT_FEDERATION_MAX_DOMAINS_PER_CRAWL = 256
     DEFAULT_FEDERATION_WORKER_POOL_SIZE = 4
     DEFAULT_FEDERATION_WORKER_QUEUE_CAPACITY = 128
     DEFAULT_FEDERATION_TASK_TIMEOUT_SECONDS = 120
+    DEFAULT_FEDERATION_SHUTDOWN_TIMEOUT_SECONDS = 3
+    DEFAULT_FEDERATION_CRAWL_COOLDOWN_SECONDS = 300
     DEFAULT_INITIAL_FEDERATION_DELAY_SECONDS = 2
     DEFAULT_FEDERATION_SEED_DOMAINS = %w[potatomesh.net potatomesh.jmrp.io mesh.qrp.ro].freeze
 
@@ -350,6 +353,16 @@ module PotatoMesh
       )
     end
 
+    # End-to-end timeout applied to each outbound federation HTTP request.
+    #
+    # @return [Integer] maximum request duration in seconds.
+    def remote_instance_request_timeout
+      fetch_positive_integer(
+        "REMOTE_INSTANCE_REQUEST_TIMEOUT",
+        DEFAULT_REMOTE_INSTANCE_REQUEST_TIMEOUT,
+      )
+    end
+
     # Limit the number of remote instances processed from a single response.
     #
     # @return [Integer] maximum entries processed per /api/instances payload.
@@ -400,6 +413,26 @@ module PotatoMesh
       )
     end
 
+    # Determine how long shutdown waits before forcing federation thread exit.
+    #
+    # @return [Integer] per-thread shutdown timeout in seconds.
+    def federation_shutdown_timeout_seconds
+      fetch_positive_integer(
+        "FEDERATION_SHUTDOWN_TIMEOUT",
+        DEFAULT_FEDERATION_SHUTDOWN_TIMEOUT_SECONDS,
+      )
+    end
+
+    # Define how long finished crawl domains remain on cooldown.
+    #
+    # @return [Integer] cooldown window in seconds.
+    def federation_crawl_cooldown_seconds
+      fetch_positive_integer(
+        "FEDERATION_CRAWL_COOLDOWN",
+        DEFAULT_FEDERATION_CRAWL_COOLDOWN_SECONDS,
+      )
+    end
+
     # Maximum acceptable age for remote node data.
     #
     # @return [Integer] seconds before remote nodes are considered stale.

web/public/assets/js/app/__tests__/chat-log-tabs.test.js

@@ -62,6 +62,22 @@ function buildModel(overrides = {}) {
   });
 }
 
+function findChannelByLabel(model, label) {
+  return model.channels.find(channel => channel.label === label);
+}
+
+function assertChannelMessages(model, { label, id, index, messageIds }) {
+  const channel = findChannelByLabel(model, label);
+  assert.ok(channel);
+  if (id instanceof RegExp) {
+    assert.match(channel.id, id);
+  } else {
+    assert.equal(channel.id, id);
+  }
+  assert.equal(channel.index, index);
+  assert.deepEqual(channel.entries.map(entry => entry.message.id), messageIds);
+}
+
 test('buildChatTabModel returns sorted nodes and channel buckets', () => {
   const model = buildModel();
   assert.equal(model.logEntries.length, 3);
@@ -75,12 +91,13 @@ test('buildChatTabModel returns sorted nodes and channel buckets', () => {
     ['recent-node', 'iso-node', 'encrypted']
   );
 
-  assert.equal(model.channels.length, 5);
+  assert.equal(model.channels.length, 6);
   assert.deepEqual(model.channels.map(channel => channel.label), [
     'EnvDefault',
     'Fallback',
     'MediumFast',
     'ShortFast',
+    '1',
     'BerlinMesh'
   ]);
 
@@ -106,11 +123,16 @@ test('buildChatTabModel returns sorted nodes and channel buckets', () => {
   assert.equal(presetChannel.id, 'channel-0-shortfast');
   assert.deepEqual(presetChannel.entries.map(entry => entry.message.id), ['primary-preset']);
 
+  const unnamedSecondaryChannel = channelByLabel['1'];
+  assert.equal(unnamedSecondaryChannel.index, 1);
+  assert.equal(unnamedSecondaryChannel.id, 'channel-1');
+  assert.deepEqual(unnamedSecondaryChannel.entries.map(entry => entry.message.id), ['iso-ts']);
+
   const secondaryChannel = channelByLabel.BerlinMesh;
   assert.equal(secondaryChannel.index, 1);
-  assert.equal(secondaryChannel.id, 'channel-secondary-berlinmesh');
-  assert.equal(secondaryChannel.entries.length, 2);
-  assert.deepEqual(secondaryChannel.entries.map(entry => entry.message.id), ['iso-ts', 'recent-alt']);
+  assert.match(secondaryChannel.id, /^channel-secondary-1-berlinmesh-[a-z0-9]+$/);
+  assert.equal(secondaryChannel.entries.length, 1);
+  assert.deepEqual(secondaryChannel.entries.map(entry => entry.message.id), ['recent-alt']);
 });
 
 test('buildChatTabModel skips channel buckets when there are no messages', () => {
@@ -272,7 +294,7 @@ test('buildChatTabModel ignores plaintext log-only entries', () => {
   assert.equal(encryptedEntries[0]?.message?.id, 'enc');
 });
 
-test('buildChatTabModel merges secondary channels with matching labels regardless of index', () => {
+test('buildChatTabModel keeps secondary channels distinct by index even with matching labels', () => {
   const primaryId = 'primary';
   const secondaryFirstId = 'secondary-one';
   const secondarySecondId = 'secondary-two';
@@ -289,62 +311,129 @@ test('buildChatTabModel merges secondary channels with matching labels regardles
   });
 
   const meshChannels = model.channels.filter(channel => channel.label === label);
-  assert.equal(meshChannels.length, 2);
+  assert.equal(meshChannels.length, 3);
 
   const primaryChannel = meshChannels.find(channel => channel.index === 0);
   assert.ok(primaryChannel);
   assert.equal(primaryChannel.entries.length, 1);
   assert.equal(primaryChannel.entries[0]?.message?.id, primaryId);
 
-  const secondaryChannel = meshChannels.find(channel => channel.index > 0);
-  assert.ok(secondaryChannel);
-  assert.equal(secondaryChannel.id, 'channel-secondary-meshtown');
-  assert.equal(secondaryChannel.index, 3);
-  assert.deepEqual(secondaryChannel.entries.map(entry => entry.message.id), [secondaryFirstId, secondarySecondId]);
+  const secondaryFirstChannel = meshChannels.find(channel => channel.index === 7);
+  assert.ok(secondaryFirstChannel);
+  assert.match(secondaryFirstChannel.id, /^channel-secondary-7-meshtown-[a-z0-9]+$/);
+  assert.deepEqual(secondaryFirstChannel.entries.map(entry => entry.message.id), [secondaryFirstId]);
+
+  const secondarySecondChannel = meshChannels.find(channel => channel.index === 3);
+  assert.ok(secondarySecondChannel);
+  assert.match(secondarySecondChannel.id, /^channel-secondary-3-meshtown-[a-z0-9]+$/);
+  assert.deepEqual(secondarySecondChannel.entries.map(entry => entry.message.id), [secondarySecondId]);
 });
 
-test('buildChatTabModel rekeys unnamed secondary buckets when a label later arrives', () => {
-  const unnamedId = 'unnamed';
-  const namedId = 'named';
-  const label = 'SideMesh';
-  const index = 4;
+test('buildChatTabModel keeps unnamed secondary buckets separate when a label later arrives', () => {
+  const scenarios = [
+    {
+      index: 4,
+      label: 'SideMesh',
+      messages: [
+        { id: 'unnamed', rx_time: NOW - 15, channel: 4 },
+        { id: 'named', rx_time: NOW - 10, channel: 4, channel_name: 'SideMesh' }
+      ],
+      namedId: /^channel-secondary-4-sidemesh-[a-z0-9]+$/,
+      namedMessages: ['named'],
+      unnamedMessages: ['unnamed']
+    },
+    {
+      index: 5,
+      label: 'MeshNorth',
+      messages: [
+        { id: 'named', rx_time: NOW - 12, channel: 5, channel_name: 'MeshNorth' },
+        { id: 'unlabeled', rx_time: NOW - 8, channel: 5 }
+      ],
+      namedId: /^channel-secondary-5-meshnorth-[a-z0-9]+$/,
+      namedMessages: ['named'],
+      unnamedMessages: ['unlabeled']
+    }
+  ];
+
+  for (const scenario of scenarios) {
+    const model = buildChatTabModel({
+      nodes: [],
+      messages: scenario.messages,
+      nowSeconds: NOW,
+      windowSeconds: WINDOW
+    });
+    const secondaryChannels = model.channels.filter(channel => channel.index === scenario.index);
+    assert.equal(secondaryChannels.length, 2);
+    assertChannelMessages(model, {
+      label: scenario.label,
+      id: scenario.namedId,
+      index: scenario.index,
+      messageIds: scenario.namedMessages
+    });
+    assertChannelMessages(model, {
+      label: String(scenario.index),
+      id: `channel-${scenario.index}`,
+      index: scenario.index,
+      messageIds: scenario.unnamedMessages
+    });
+  }
+});
+
+test('buildChatTabModel keeps same-index channels with different names in separate tabs', () => {
   const model = buildChatTabModel({
     nodes: [],
     messages: [
-      { id: unnamedId, rx_time: NOW - 15, channel: index },
-      { id: namedId, rx_time: NOW - 10, channel: index, channel_name: label }
+      { id: 'public-msg', rx_time: NOW - 12, channel: 1, channel_name: 'PUBLIC' },
+      { id: 'berlin-msg', rx_time: NOW - 8, channel: 1, channel_name: 'BerlinMesh' }
     ],
     nowSeconds: NOW,
     windowSeconds: WINDOW
   });
 
-  const secondaryChannels = model.channels.filter(channel => channel.index === index);
-  assert.equal(secondaryChannels.length, 1);
-  const [secondaryChannel] = secondaryChannels;
-  assert.equal(secondaryChannel.id, 'channel-secondary-sidemesh');
-  assert.equal(secondaryChannel.label, label);
-  assert.deepEqual(secondaryChannel.entries.map(entry => entry.message.id), [unnamedId, namedId]);
+  assertChannelMessages(model, {
+    label: 'PUBLIC',
+    id: /^channel-secondary-1-public-[a-z0-9]+$/,
+    index: 1,
+    messageIds: ['public-msg']
+  });
+  assertChannelMessages(model, {
+    label: 'BerlinMesh',
+    id: /^channel-secondary-1-berlinmesh-[a-z0-9]+$/,
+    index: 1,
+    messageIds: ['berlin-msg']
+  });
 });
 
-test('buildChatTabModel merges unlabeled secondary messages into existing named buckets by index', () => {
-  const namedId = 'named';
-  const unlabeledId = 'unlabeled';
-  const label = 'MeshNorth';
-  const index = 5;
+test('buildChatTabModel keeps same-index slug-colliding labels on distinct tab ids', () => {
   const model = buildChatTabModel({
     nodes: [],
     messages: [
-      { id: namedId, rx_time: NOW - 12, channel: index, channel_name: label },
-      { id: unlabeledId, rx_time: NOW - 8, channel: index }
+      { id: 'foo-space', rx_time: NOW - 10, channel: 1, channel_name: 'Foo Bar' },
+      { id: 'foo-dash', rx_time: NOW - 8, channel: 1, channel_name: 'Foo-Bar' }
     ],
     nowSeconds: NOW,
     windowSeconds: WINDOW
   });
 
-  const secondaryChannels = model.channels.filter(channel => channel.index === index);
-  assert.equal(secondaryChannels.length, 1);
-  const [secondaryChannel] = secondaryChannels;
-  assert.equal(secondaryChannel.id, 'channel-secondary-meshnorth');
-  assert.equal(secondaryChannel.label, label);
-  assert.deepEqual(secondaryChannel.entries.map(entry => entry.message.id), [namedId, unlabeledId]);
+  const fooSpaceChannel = findChannelByLabel(model, 'Foo Bar');
+  const fooDashChannel = findChannelByLabel(model, 'Foo-Bar');
+  assert.ok(fooSpaceChannel);
+  assert.ok(fooDashChannel);
+  assert.match(fooSpaceChannel.id, /^channel-secondary-1-foo-bar-[a-z0-9]+$/);
+  assert.match(fooDashChannel.id, /^channel-secondary-1-foo-bar-[a-z0-9]+$/);
+  assert.notEqual(fooSpaceChannel.id, fooDashChannel.id);
+});
+
+test('buildChatTabModel falls back to hashed id for unsluggable secondary labels', () => {
+  const model = buildChatTabModel({
+    nodes: [],
+    messages: [{ id: 'hash-fallback', rx_time: NOW - 5, channel: 2, channel_name: '###' }],
+    nowSeconds: NOW,
+    windowSeconds: WINDOW
+  });
+  const channel = findChannelByLabel(model, '###');
+  assert.ok(channel);
+  assert.equal(channel.index, 2);
+  assert.ok(channel.id.startsWith('channel-secondary-2-'));
+  assert.ok(channel.id.length > 'channel-secondary-2-'.length);
 });

web/public/assets/js/app/__tests__/node-page.test.js

@@ -154,12 +154,14 @@ test('additional format helpers provide table friendly output', () => {
         channel_name: 'Primary',
         node: { short_name: 'SRCE', role: 'ROUTER', node_id: '!src' },
       },
+      { text: ' GAA= ', encrypted: true, rx_time: 1_700_000_405 },
       { emoji: '😊', rx_time: 1_700_000_401 },
     ],
     renderShortHtml,
     nodeContext,
   );
   assert.equal(messagesHtml.includes('hello'), true);
+  assert.equal(messagesHtml.includes('GAA='), false);
   assert.equal(messagesHtml.includes('😊'), true);
   assert.match(messagesHtml, /\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}\]\[868\]/);
   assert.equal(messagesHtml.includes('[868]'), true);

web/public/assets/js/app/chat-log-tabs.js

@@ -20,7 +20,7 @@ import { extractModemMetadata } from './node-modem-metadata.js';
  * Highest channel index that should be represented within the tab view.
  * @type {number}
  */
-export const MAX_CHANNEL_INDEX = 9;
+export const MAX_CHANNEL_INDEX = 255;
 
 /**
  * Discrete event types that can appear in the chat activity log.
@@ -245,28 +245,12 @@ export function buildChatTabModel({
       modemPreset,
       envFallbackLabel: primaryChannelEnvLabel
     });
-    const nameBucketKey = safeIndex > 0 ? buildSecondaryNameBucketKey(labelInfo) : null;
+    const nameBucketKey = safeIndex > 0 ? buildSecondaryNameBucketKey(safeIndex, labelInfo) : null;
     const primaryBucketKey = safeIndex === 0 && labelInfo.label !== '0' ? buildPrimaryBucketKey(labelInfo.label) : '0';
 
-    let bucketKey = safeIndex === 0 ? primaryBucketKey : nameBucketKey ?? String(safeIndex);
+    const bucketKey = safeIndex === 0 ? primaryBucketKey : nameBucketKey ?? String(safeIndex);
     let bucket = channelBuckets.get(bucketKey);
 
-    if (!bucket && safeIndex > 0) {
-      const existingBucketKey = findExistingBucketKeyByIndex(channelBuckets, safeIndex);
-      if (existingBucketKey) {
-        bucketKey = existingBucketKey;
-        bucket = channelBuckets.get(existingBucketKey);
-      }
-    }
-
-    if (bucket && nameBucketKey && bucket.key !== nameBucketKey) {
-      channelBuckets.delete(bucket.key);
-      bucket.key = nameBucketKey;
-      bucket.id = buildChannelTabId(nameBucketKey);
-      channelBuckets.set(nameBucketKey, bucket);
-      bucketKey = nameBucketKey;
-    }
-
     if (!bucket) {
       bucket = {
         key: bucketKey,
@@ -569,43 +553,34 @@ function buildPrimaryBucketKey(primaryChannelLabel) {
   return '0';
 }
 
-function buildSecondaryNameBucketKey(labelInfo) {
+function buildSecondaryNameBucketKey(index, labelInfo) {
   const label = labelInfo?.label ?? null;
   const priority = labelInfo?.priority ?? CHANNEL_LABEL_PRIORITY.INDEX;
-  if (priority !== CHANNEL_LABEL_PRIORITY.NAME || !label) {
+  const safeIndex = Number.isFinite(index) ? Math.max(0, Math.trunc(index)) : 0;
+  if (safeIndex <= 0 || priority !== CHANNEL_LABEL_PRIORITY.NAME || !label) {
     return null;
   }
   const trimmedLabel = label.trim().toLowerCase();
   if (!trimmedLabel.length) {
     return null;
   }
-  return `secondary::${trimmedLabel}`;
-}
-
-function findExistingBucketKeyByIndex(channelBuckets, targetIndex) {
-  if (!channelBuckets || !Number.isFinite(targetIndex) || targetIndex <= 0) {
-    return null;
-  }
-  const normalizedTarget = Math.trunc(targetIndex);
-  for (const [key, bucket] of channelBuckets.entries()) {
-    if (!bucket || !Number.isFinite(bucket.index)) {
-      continue;
-    }
-    if (Math.trunc(bucket.index) !== normalizedTarget) {
-      continue;
-    }
-    if (bucket.index === 0) {
-      continue;
-    }
-    return key;
-  }
-  return null;
+  return `secondary::${safeIndex}::${trimmedLabel}`;
 }
 
 function buildChannelTabId(bucketKey) {
   if (bucketKey === '0') {
     return 'channel-0';
   }
+  const secondaryParts = /^secondary::(\d+)::(.+)$/.exec(String(bucketKey));
+  if (secondaryParts) {
+    const secondaryIndex = secondaryParts[1];
+    const secondaryLabelSlug = slugify(secondaryParts[2]);
+    const secondaryHash = hashChannelKey(bucketKey);
+    if (secondaryLabelSlug) {
+      return `channel-secondary-${secondaryIndex}-${secondaryLabelSlug}-${secondaryHash}`;
+    }
+    return `channel-secondary-${secondaryIndex}-${secondaryHash}`;
+  }
   const slug = slugify(bucketKey);
   if (slug) {
     if (slug !== '0') {

web/public/assets/js/app/main.js

@@ -2910,6 +2910,14 @@ export function initializeApp(config) {
    * @returns {HTMLElement} Chat log element.
    */
   function createMessageChatEntry(m) {
+    let plainText = '';
+    if (m?.text != null) {
+      plainText = String(m.text).trim();
+    }
+    if (m?.encrypted && plainText === 'GAA=') {
+      return null;
+    }
+
     const div = document.createElement('div');
     const tsSeconds = resolveTimestampSeconds(
       m.rx_time ?? m.rxTime,
@@ -3141,18 +3149,28 @@ export function initializeApp(config) {
     }
     const getDivider = createDateDividerFactory();
     const limitedEntries = entries.slice(Math.max(entries.length - CHAT_LIMIT, 0));
+    let renderedEntries = 0;
     for (const entry of limitedEntries) {
       if (!entry || typeof entry.ts !== 'number') {
         continue;
       }
+      if (typeof renderEntry !== 'function') {
+        continue;
+      }
+      const node = renderEntry(entry);
+      if (!node) {
+        continue;
+      }
       const divider = getDivider(entry.ts);
       if (divider) fragment.appendChild(divider);
-      if (typeof renderEntry === 'function') {
-        const node = renderEntry(entry);
-        if (node) {
-          fragment.appendChild(node);
-        }
-      }
+      fragment.appendChild(node);
+      renderedEntries += 1;
+    }
+    if (renderedEntries === 0 && emptyLabel) {
+      const empty = document.createElement('p');
+      empty.className = 'chat-empty';
+      empty.textContent = emptyLabel;
+      fragment.appendChild(empty);
     }
     return fragment;
   }

web/public/assets/js/app/node-page.js

@@ -2056,6 +2056,7 @@ function renderMessages(messages, renderShortHtml, node) {
       if (!message || typeof message !== 'object') return null;
       const text = stringOrNull(message.text) || stringOrNull(message.emoji);
       if (!text) return null;
+      if (message.encrypted && String(text).trim() === 'GAA=') return null;
 
       const timestamp = formatMessageTimestamp(message.rx_time, message.rx_iso);
       const metadata = extractChatMessageMetadata(message);

web/spec/app_spec.rb

@@ -26,6 +26,8 @@ require "socket"
 RSpec.describe "Potato Mesh Sinatra app" do
   let(:app) { Sinatra::Application }
   let(:application_class) { PotatoMesh::Application }
+  INSERT_NODE_WITH_LAST_HEARD_SQL = "INSERT INTO nodes(node_id, num, last_heard, first_heard) VALUES (?,?,?,?)".freeze
+  SELECT_NODE_LAST_HEARD_SQL = "SELECT last_heard FROM nodes WHERE node_id = ?".freeze
 
   describe "configuration" do
     it "sets the default HTTP port to the baked-in value" do
@@ -406,7 +408,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
     it "stores and clears the initial federation thread" do
       delay = 3
       allow(PotatoMesh::Config).to receive(:initial_federation_delay_seconds).and_return(delay)
-      expect(Kernel).to receive(:sleep).with(delay)
+      expect(app).to receive(:federation_sleep_with_shutdown).with(delay).and_return(true)
       expect(app).to receive(:announce_instance_to_all_domains)
       allow(Thread).to receive(:new) do |&block|
         dummy_thread.block = block
@@ -926,7 +928,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
 
         with_db do |db|
           db.execute(
-            "INSERT INTO nodes(node_id, num, last_heard, first_heard) VALUES (?,?,?,?)",
+            INSERT_NODE_WITH_LAST_HEARD_SQL,
             [node_id, node_num, rx_time - 120, rx_time - 180],
           )
 
@@ -2865,7 +2867,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
         count = db.get_first_value("SELECT COUNT(*) FROM nodes WHERE node_id = ?", [node["node_id"]])
         expect(count).to eq(1)
 
-        last_heard = db.get_first_value("SELECT last_heard FROM nodes WHERE node_id = ?", [node["node_id"]])
+        last_heard = db.get_first_value(SELECT_NODE_LAST_HEARD_SQL, [node["node_id"]])
         expect(last_heard).to eq(expected_last_heard(node))
       end
     end
@@ -2928,6 +2930,26 @@ RSpec.describe "Potato Mesh Sinatra app" do
   end
 
   describe "POST /api/messages" do
+    SELECT_MESSAGE_ENCRYPTED_SQL = "SELECT encrypted FROM messages WHERE id = ?".freeze
+    SELECT_NEIGHBOR_COUNT_BY_NODE_SQL = "SELECT COUNT(*) FROM neighbors WHERE node_id = ?".freeze
+    NODE_INFO_LONG_NAME = "Node Info".freeze
+    FIRST_MESSAGE_INGESTOR_ID = "!1111aaaa".freeze
+    SHARED_TEST_INGESTOR_ID = "!aaaa1111".freeze
+    DEADBEEF_NODE_ID = "!deadbeef".freeze
+    NEIGHBOR_EMPTY_UPDATE_ROOT_ID = "!cafed00d".freeze
+    NEIGHBOR_ROOT_ID = "!1a2b3c01".freeze
+    NEIGHBOR_PRIMARY_ID = "!1a2b3c02".freeze
+    NEIGHBOR_SNR_CLEAR_ROOT_ID = "!1a2b3c10".freeze
+    NEIGHBOR_SNR_CLEAR_PEER_ID = "!1a2b3c11".freeze
+    NEIGHBOR_CHUNK_ROOT_ID = "!1a2b3c30".freeze
+
+    def post_twice_for_ingestor(endpoint, first_payload, second_payload)
+      post endpoint, first_payload.to_json, auth_headers
+      expect(last_response).to be_ok
+      post endpoint, second_payload.to_json, auth_headers
+      expect(last_response).to be_ok
+    end
+
     it "persists messages from fixture data" do
       import_nodes_fixture
       import_messages_fixture
@@ -3011,6 +3033,36 @@ RSpec.describe "Potato Mesh Sinatra app" do
       expect(reaction_row["emoji"]).to eq("🔥")
     end
 
+    it "stores message ingestor and preserves the first reporter" do
+      first_payload = {
+        "id" => 77_001,
+        "rx_time" => reference_time.to_i - 10,
+        "from_id" => "!ingmsg01",
+        "channel" => 0,
+        "portnum" => "TEXT_MESSAGE_APP",
+        "text" => "first reporter",
+        "ingestor" => FIRST_MESSAGE_INGESTOR_ID,
+      }
+      second_payload = first_payload.merge(
+        "text" => "updated text",
+        "ingestor" => "!2222bbbb",
+      )
+
+      post_twice_for_ingestor("/api/messages", first_payload, second_payload)
+
+      with_db(readonly: true) do |db|
+        db.results_as_hash = true
+        row = db.get_first_row("SELECT text, ingestor FROM messages WHERE id = ?", [first_payload["id"]])
+        expect(row["text"]).to eq("updated text")
+        expect(row["ingestor"]).to eq(FIRST_MESSAGE_INGESTOR_ID)
+      end
+
+      get "/api/messages?limit=10"
+      expect(last_response).to be_ok
+      row = JSON.parse(last_response.body).find { |entry| entry["id"] == first_payload["id"] }
+      expect(row["ingestor"]).to eq(FIRST_MESSAGE_INGESTOR_ID)
+    end
+
     it "creates hidden nodes for unknown message senders" do
       payload = {
         "id" => 9_999,
@@ -3075,7 +3127,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
         "id" => 10_001,
         "rx_time" => reference_time.to_i,
         "from_id" => "!cafef00d",
-        "to_id" => "!deadbeef",
+        "to_id" => DEADBEEF_NODE_ID,
         "channel" => 0,
         "portnum" => "TEXT_MESSAGE_APP",
         "text" => "Spec participant placeholder",
@@ -3092,12 +3144,12 @@ RSpec.describe "Potato Mesh Sinatra app" do
           <<~SQL,
           SELECT node_id, num, short_name, long_name, role, last_heard, first_heard
           FROM nodes
-          WHERE node_id IN ("!cafef00d", "!deadbeef")
+          WHERE node_id IN ("!cafef00d", "#{DEADBEEF_NODE_ID}")
           ORDER BY node_id
         SQL
         )
 
-        expect(rows.map { |row| row["node_id"] }).to contain_exactly("!cafef00d", "!deadbeef")
+        expect(rows.map { |row| row["node_id"] }).to contain_exactly("!cafef00d", DEADBEEF_NODE_ID)
         rows.each do |row|
           expect(row["num"]).to be_an(Integer)
           expect(row["role"]).to eq("CLIENT_HIDDEN")
@@ -3296,6 +3348,30 @@ RSpec.describe "Potato Mesh Sinatra app" do
         end
       end
 
+      it "stores position ingestor and preserves the first reporter" do
+        first_payload = {
+          "id" => 19_001,
+          "node_id" => "!ingpos01",
+          "rx_time" => reference_time.to_i - 80,
+          "latitude" => 52.1,
+          "longitude" => 13.2,
+          "ingestor" => SHARED_TEST_INGESTOR_ID,
+        }
+        second_payload = first_payload.merge(
+          "latitude" => 53.3,
+          "ingestor" => "!bbbb2222",
+        )
+
+        post_twice_for_ingestor("/api/positions", first_payload, second_payload)
+
+        with_db(readonly: true) do |db|
+          db.results_as_hash = true
+          row = db.get_first_row("SELECT latitude, ingestor FROM positions WHERE id = ?", [first_payload["id"]])
+          expect_same_value(row["latitude"], 53.3)
+          expect(row["ingestor"]).to eq(SHARED_TEST_INGESTOR_ID)
+        end
+      end
+
       it "fills first_heard when updating an existing node without one" do
         node_id = "!specposfh"
         rx_time = reference_time.to_i - 90
@@ -3438,6 +3514,48 @@ RSpec.describe "Potato Mesh Sinatra app" do
         end
       end
 
+      it "does not update existing neighbor last_heard from third-party neighbor reports" do
+        reporter_id = "!abc123ef"
+        existing_neighbor_id = "!00ff0011"
+        prior_last_heard = reference_time.to_i - 4 * 60 * 60
+        rx_time = reference_time.to_i - 60 * 60
+        neighbor_rx_time = rx_time - 120
+
+        with_db do |db|
+          db.execute(
+            INSERT_NODE_WITH_LAST_HEARD_SQL,
+            [reporter_id, 0xabc123ef, prior_last_heard, prior_last_heard],
+          )
+          db.execute(
+            INSERT_NODE_WITH_LAST_HEARD_SQL,
+            [existing_neighbor_id, 0x00ff0011, prior_last_heard, prior_last_heard],
+          )
+        end
+
+        payload = {
+          "node_id" => reporter_id,
+          "node_num" => 0xabc123ef,
+          "rx_time" => rx_time,
+          "neighbors" => [
+            { "node_id" => existing_neighbor_id, "snr" => -7.5, "rx_time" => neighbor_rx_time },
+          ],
+        }
+
+        post "/api/neighbors", payload.to_json, auth_headers
+
+        expect(last_response).to be_ok
+        expect(JSON.parse(last_response.body)).to eq("status" => "ok")
+
+        with_db(readonly: true) do |db|
+          db.results_as_hash = true
+          reporter_row = db.get_first_row(SELECT_NODE_LAST_HEARD_SQL, [reporter_id])
+          neighbor_row = db.get_first_row(SELECT_NODE_LAST_HEARD_SQL, [existing_neighbor_id])
+
+          expect(reporter_row["last_heard"]).to eq(rx_time)
+          expect(neighbor_row["last_heard"]).to eq(prior_last_heard)
+        end
+      end
+
       it "handles broadcasts with no neighbors" do
         rx_time = reference_time.to_i - 60
         payload = {
@@ -3469,6 +3587,127 @@ RSpec.describe "Potato Mesh Sinatra app" do
         expect(JSON.parse(last_response.body)).to be_empty
       end
 
+      it "removes stored neighbors when a later packet contains no neighbors" do
+        seed_payload = {
+          "node_id" => NEIGHBOR_EMPTY_UPDATE_ROOT_ID,
+          "rx_time" => reference_time.to_i - 50,
+          "neighbors" => [
+            { "node_id" => DEADBEEF_NODE_ID, "snr" => -2.0 },
+          ],
+          "ingestor" => SHARED_TEST_INGESTOR_ID,
+        }
+        empty_payload = {
+          "node_id" => NEIGHBOR_EMPTY_UPDATE_ROOT_ID,
+          "rx_time" => reference_time.to_i - 10,
+          "neighbors" => [],
+          "ingestor" => "!bbbb2222",
+        }
+
+        post "/api/neighbors", seed_payload.to_json, auth_headers
+        expect(last_response).to be_ok
+        post "/api/neighbors", empty_payload.to_json, auth_headers
+        expect(last_response).to be_ok
+
+        with_db(readonly: true) do |db|
+          remaining = db.get_first_value(SELECT_NEIGHBOR_COUNT_BY_NODE_SQL, [NEIGHBOR_EMPTY_UPDATE_ROOT_ID])
+          expect(remaining).to eq(0)
+        end
+      end
+
+      it "stores neighbor ingestor and preserves the first reporter per tuple" do
+        base = {
+          "node_id" => NEIGHBOR_ROOT_ID,
+          "rx_time" => reference_time.to_i - 45,
+          "neighbors" => [
+            { "node_id" => NEIGHBOR_PRIMARY_ID, "snr" => -1.5 },
+            { "node_id" => "!1a2b3c03", "snr" => -2.5 },
+          ],
+          "ingestor" => "!aaaa9999",
+        }
+        update = {
+          "node_id" => NEIGHBOR_ROOT_ID,
+          "rx_time" => reference_time.to_i - 30,
+          "neighbors" => [
+            { "node_id" => NEIGHBOR_PRIMARY_ID, "snr" => -0.5 },
+          ],
+          "ingestor" => "!bbbb8888",
+        }
+
+        post_twice_for_ingestor("/api/neighbors", base, update)
+
+        with_db(readonly: true) do |db|
+          db.results_as_hash = true
+          rows = db.execute("SELECT neighbor_id, snr, ingestor FROM neighbors WHERE node_id = ? ORDER BY neighbor_id", [NEIGHBOR_ROOT_ID])
+          expect(rows.size).to eq(1)
+          expect(rows.first["neighbor_id"]).to eq(NEIGHBOR_PRIMARY_ID)
+          expect_same_value(rows.first["snr"], -0.5)
+          expect(rows.first["ingestor"]).to eq("!aaaa9999")
+        end
+      end
+
+      it "clears stored neighbor snr when an updated entry omits snr" do
+        initial = {
+          "node_id" => NEIGHBOR_SNR_CLEAR_ROOT_ID,
+          "rx_time" => reference_time.to_i - 40,
+          "neighbors" => [
+            { "node_id" => NEIGHBOR_SNR_CLEAR_PEER_ID, "snr" => -3.25 },
+          ],
+        }
+        update = {
+          "node_id" => NEIGHBOR_SNR_CLEAR_ROOT_ID,
+          "rx_time" => reference_time.to_i - 20,
+          "neighbors" => [
+            { "node_id" => NEIGHBOR_SNR_CLEAR_PEER_ID },
+          ],
+        }
+
+        post "/api/neighbors", initial.to_json, auth_headers
+        expect(last_response).to be_ok
+        post "/api/neighbors", update.to_json, auth_headers
+        expect(last_response).to be_ok
+
+        with_db(readonly: true) do |db|
+          db.results_as_hash = true
+          row = db.get_first_row(
+            "SELECT snr, rx_time FROM neighbors WHERE node_id = ? AND neighbor_id = ?",
+            [NEIGHBOR_SNR_CLEAR_ROOT_ID, NEIGHBOR_SNR_CLEAR_PEER_ID],
+          )
+          expect(row["snr"]).to be_nil
+          expect(row["rx_time"]).to eq(update["rx_time"])
+        end
+      end
+
+      it "removes stale neighbors in chunked deletes" do
+        initial_neighbors = Array.new(1_100) do |i|
+          { "node_id" => format("!%08x", 0x2000_0000 + i), "snr" => -2.0 }
+        end
+        initial = {
+          "node_id" => NEIGHBOR_CHUNK_ROOT_ID,
+          "rx_time" => reference_time.to_i - 35,
+          "neighbors" => initial_neighbors,
+        }
+        update = {
+          "node_id" => NEIGHBOR_CHUNK_ROOT_ID,
+          "rx_time" => reference_time.to_i - 25,
+          "neighbors" => [
+            { "node_id" => "!20000000", "snr" => -1.0 },
+          ],
+        }
+
+        post "/api/neighbors", initial.to_json, auth_headers
+        expect(last_response).to be_ok
+        post "/api/neighbors", update.to_json, auth_headers
+        expect(last_response).to be_ok
+
+        with_db(readonly: true) do |db|
+          count = db.get_first_value(
+            SELECT_NEIGHBOR_COUNT_BY_NODE_SQL,
+            [NEIGHBOR_CHUNK_ROOT_ID],
+          )
+          expect(count).to eq(1)
+        end
+      end
+
       it "returns 400 when more than 1000 neighbor packets are provided" do
         payload = Array.new(1001) do |i|
           { "node_id" => format("!%08x", i), "rx_time" => reference_time.to_i - i }
@@ -3484,6 +3723,30 @@ RSpec.describe "Potato Mesh Sinatra app" do
           expect(count).to eq(0)
         end
       end
+
+      it "handles large neighbor lists without SQLite bind overflows" do
+        neighbors = Array.new(1_100) do |i|
+          { "node_id" => format("!%08x", 0x1000_0000 + i), "snr" => -1.0 }
+        end
+        payload = {
+          "node_id" => "!1a2b3c20",
+          "rx_time" => reference_time.to_i - 15,
+          "neighbors" => neighbors,
+        }
+
+        post "/api/neighbors", payload.to_json, auth_headers
+
+        expect(last_response).to be_ok
+        expect(JSON.parse(last_response.body)).to eq("status" => "ok")
+
+        with_db(readonly: true) do |db|
+          count = db.get_first_value(
+            SELECT_NEIGHBOR_COUNT_BY_NODE_SQL,
+            ["!1a2b3c20"],
+          )
+          expect(count).to eq(1_100)
+        end
+      end
     end
 
     describe "POST /api/telemetry" do
@@ -3641,6 +3904,27 @@ RSpec.describe "Potato Mesh Sinatra app" do
         expect(JSON.parse(last_response.body)).to eq("error" => "invalid JSON")
       end
 
+      it "stores telemetry ingestor and preserves the first reporter" do
+        payload = {
+          "id" => 23_001,
+          "node_id" => "!ingtel01",
+          "rx_time" => reference_time.to_i - 70,
+          "telemetry" => { "deviceMetrics" => { "batteryLevel" => 90 } },
+          "battery_level" => 90,
+          "ingestor" => "!1111bbbb",
+        }
+        updated = payload.merge("battery_level" => 80, "ingestor" => "!2222cccc")
+
+        post_twice_for_ingestor("/api/telemetry", payload, updated)
+
+        with_db(readonly: true) do |db|
+          db.results_as_hash = true
+          row = db.get_first_row("SELECT battery_level, ingestor FROM telemetry WHERE id = ?", [payload["id"]])
+          expect_same_value(row["battery_level"], 80.0)
+          expect(row["ingestor"]).to eq("!1111bbbb")
+        end
+      end
+
       it "returns 400 when more than 1000 telemetry packets are provided" do
         payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } }
 
@@ -3760,6 +4044,28 @@ RSpec.describe "Potato Mesh Sinatra app" do
         expect(JSON.parse(last_response.body)).to eq("error" => "invalid JSON")
       end
 
+      it "stores trace ingestor and preserves the first reporter" do
+        payload = {
+          "id" => 31_001,
+          "request_id" => 77,
+          "src" => 0x10000001,
+          "dest" => 0x10000002,
+          "rx_time" => reference_time.to_i - 50,
+          "hops" => [0x10000001, 0x10000002],
+          "ingestor" => "!aaaa0001",
+        }
+        update = payload.merge("snr" => 7.5, "ingestor" => "!bbbb0002")
+
+        post_twice_for_ingestor("/api/traces", payload, update)
+
+        with_db(readonly: true) do |db|
+          db.results_as_hash = true
+          row = db.get_first_row("SELECT snr, ingestor FROM traces WHERE id = ?", [payload["id"]])
+          expect_same_value(row["snr"], 7.5)
+          expect(row["ingestor"]).to eq("!aaaa0001")
+        end
+      end
+
       it "returns 400 when more than 1000 traces are provided" do
         payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } }
 
@@ -3938,7 +4244,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
         expect(row["encrypted"]).to eq(encrypted_b64)
 
         node_row = db.get_first_row(
-          "SELECT last_heard FROM nodes WHERE node_id = ?",
+          SELECT_NODE_LAST_HEARD_SQL,
           [sender_id],
         )
 
@@ -3983,7 +4289,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
       expect(node_entry["to_id"]).to eq(receiver_id)
     end
 
-    it "decrypts encrypted messages when the PSK is configured" do
+    it "keeps encrypted text-port messages even when the PSK is configured" do
       psk_b64 = "Nmh7EooP2Tsc+7pvPwXLcEDDuYhk+fBo2GLnbA1Y1sg="
       previous_psk = ENV["MESHTASTIC_PSK_B64"]
       ENV["MESHTASTIC_PSK_B64"] = psk_b64
@@ -4011,9 +4317,9 @@ RSpec.describe "Potato Mesh Sinatra app" do
             [payload["packet_id"]],
           )
 
-          expect(row["text"]).to eq("Nabend")
-          expect(row["encrypted"]).to be_nil
-          expect(row["channel_name"]).to eq("BerlinMesh")
+          expect(row["text"]).to be_nil
+          expect(row["encrypted"]).to eq("Q1R7tgI5yXzMXu/3")
+          expect(row["channel_name"]).to be_nil
         end
       ensure
         if previous_psk.nil?
@@ -4388,7 +4694,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
       with_db(readonly: true) do |db|
         db.results_as_hash = true
         row = db.get_first_row(
-          "SELECT encrypted FROM messages WHERE id = ?",
+          SELECT_MESSAGE_ENCRYPTED_SQL,
           [900_005],
         )
 
@@ -4452,7 +4758,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
       with_db(readonly: true) do |db|
         db.results_as_hash = true
         row = db.get_first_row(
-          "SELECT encrypted FROM messages WHERE id = ?",
+          SELECT_MESSAGE_ENCRYPTED_SQL,
           [900_006],
         )
 
@@ -4636,22 +4942,28 @@ RSpec.describe "Potato Mesh Sinatra app" do
           "SELECT hop_index, node_id FROM trace_hops WHERE trace_id = ? ORDER BY hop_index",
           [900_004],
         )
+        hop_node_ids = hop_rows.map do |row|
+          if row.is_a?(Hash)
+            row["node_id"] || row[:node_id] || row[1]
+          else
+            row[1]
+          end
+        end
 
         expect(trace_row["id"]).to eq(900_004)
         expect(trace_row["dest"]).to eq(4)
-        expect(hop_rows.map { |row| row[1] }).to eq([1, 2, 3, 4])
+        expect(hop_node_ids).to eq([1, 2, 3, 4])
       end
     end
 
-    it "overwrites encrypted messages when decrypted text arrives" do
+    it "keeps encrypted messages when decrypted payload resolves to text portnum" do
       encrypted_payload = Base64.strict_encode64("cipher".b)
-      decrypted_text = "decoded"
       allow(PotatoMesh::Application).to receive(:decrypt_meshtastic_message).and_return(
         nil,
         {
           portnum: 1,
           payload: "plain".b,
-          text: decrypted_text,
+          text: "decoded",
           channel_name: nil,
         },
       )
@@ -4687,11 +4999,212 @@ RSpec.describe "Potato Mesh Sinatra app" do
           [910_001],
         )
 
-        expect(row["text"]).to eq(decrypted_text)
-        expect(row["encrypted"]).to be_nil
+        expect(row["text"]).to be_nil
+        expect(row["encrypted"]).to eq(encrypted_payload)
       end
     end
 
+    it "stores decoded node data when decrypting nodeinfo payloads" do
+      payload_bytes = "nodeinfo".b
+      encoded_payload = Base64.strict_encode64(payload_bytes)
+      node_payload = {
+        "id" => "!7c5b0920",
+        "num" => 2_085_057_824,
+        "last_heard" => reference_time.to_i,
+        "user" => {
+          "short_name" => "NODE",
+          "long_name" => NODE_INFO_LONG_NAME,
+          "hw_model" => "TBEAM",
+          "role" => "CLIENT",
+        },
+      }
+
+      allow(PotatoMesh::Application).to receive(:decrypt_meshtastic_message).and_return(
+        {
+          portnum: 4,
+          payload: payload_bytes,
+          text: nil,
+          channel_name: nil,
+        },
+      )
+      allow(PotatoMesh::App::Meshtastic::PayloadDecoder).to receive(:decode).and_return(
+        {
+          "type" => "NODEINFO_APP",
+          "payload" => node_payload,
+        },
+      )
+
+      with_db do |db|
+        PotatoMesh::Application.insert_message(
+          db,
+          {
+            "packet_id" => 900_008,
+            "rx_time" => reference_time.to_i,
+            "rx_iso" => reference_time.utc.iso8601,
+            "from_id" => "!7c5b0920",
+            "encrypted" => encoded_payload,
+          },
+        )
+      end
+
+      with_db(readonly: true) do |db|
+        db.results_as_hash = true
+        row = db.get_first_row(
+          "SELECT node_id, short_name, long_name, hw_model FROM nodes WHERE node_id = ?",
+          ["!7c5b0920"],
+        )
+
+        expect(row["node_id"]).to eq("!7c5b0920")
+        expect(row["short_name"]).to eq("NODE")
+        expect(row["long_name"]).to eq(NODE_INFO_LONG_NAME)
+        expect(row["hw_model"]).to eq("TBEAM")
+      end
+    end
+
+    it "keeps encrypted payloads when decoded nodeinfo payload is invalid" do
+      payload_bytes = "nodeinfo".b
+      encoded_payload = Base64.strict_encode64(payload_bytes)
+
+      allow(PotatoMesh::Application).to receive(:decrypt_meshtastic_message).and_return(
+        {
+          portnum: 4,
+          payload: payload_bytes,
+          text: nil,
+          channel_name: nil,
+        },
+      )
+      allow(PotatoMesh::App::Meshtastic::PayloadDecoder).to receive(:decode).and_return(
+        {
+          "type" => "NODEINFO_APP",
+          "payload" => {},
+        },
+      )
+
+      with_db do |db|
+        PotatoMesh::Application.insert_message(
+          db,
+          {
+            "packet_id" => 900_009,
+            "rx_time" => reference_time.to_i,
+            "rx_iso" => reference_time.utc.iso8601,
+            "from_id" => "!7c5b0920",
+            "encrypted" => encoded_payload,
+          },
+        )
+      end
+
+      with_db(readonly: true) do |db|
+        db.results_as_hash = true
+        row = db.get_first_row(
+          SELECT_MESSAGE_ENCRYPTED_SQL,
+          [900_009],
+        )
+
+        expect(row["encrypted"]).to eq(encoded_payload)
+      end
+    end
+
+    it "keeps encrypted payloads when decoded nodeinfo payload lacks identifying user fields" do
+      payload_bytes = "nodeinfo".b
+      encoded_payload = Base64.strict_encode64(payload_bytes)
+
+      allow(PotatoMesh::Application).to receive(:decrypt_meshtastic_message).and_return(
+        {
+          portnum: 4,
+          payload: payload_bytes,
+          text: nil,
+          channel_name: nil,
+        },
+      )
+      allow(PotatoMesh::App::Meshtastic::PayloadDecoder).to receive(:decode).and_return(
+        {
+          "type" => "NODEINFO_APP",
+          "payload" => { "id" => "!7c5b0920", "num" => 2_085_057_824 },
+        },
+      )
+
+      with_db do |db|
+        PotatoMesh::Application.insert_message(
+          db,
+          {
+            "packet_id" => 900_010,
+            "rx_time" => reference_time.to_i,
+            "rx_iso" => reference_time.utc.iso8601,
+            "from_id" => "!7c5b0920",
+            "encrypted" => encoded_payload,
+          },
+        )
+      end
+
+      with_db(readonly: true) do |db|
+        db.results_as_hash = true
+        row = db.get_first_row(
+          SELECT_MESSAGE_ENCRYPTED_SQL,
+          [900_010],
+        )
+
+        expect(row["encrypted"]).to eq(encoded_payload)
+      end
+    end
+
+    it "normalizes snake_case decoded nodeinfo payloads" do
+      payload = {
+        "user" => {
+          "short_name" => "NODE",
+          "long_name" => NODE_INFO_LONG_NAME,
+          "hw_model" => "TBEAM",
+          "public_key" => "pk",
+          "is_unmessagable" => true,
+        },
+        "device_metrics" => {
+          "battery_level" => 87,
+          "channel_utilization" => 12.5,
+          "air_util_tx" => 2.75,
+          "uptime_seconds" => 1200,
+        },
+        "position" => {
+          "precision_bits" => 13,
+          "location_source" => "LOC_MANUAL",
+        },
+        "last_heard" => reference_time.to_i,
+        "hops_away" => 2,
+        "is_favorite" => true,
+        "hw_model" => "TBEAM",
+      }
+
+      normalized = PotatoMesh::Application.send(:normalize_decrypted_nodeinfo_payload, payload)
+
+      expect(normalized.dig("user", "shortName")).to eq("NODE")
+      expect(normalized.dig("user", "longName")).to eq(NODE_INFO_LONG_NAME)
+      expect(normalized.dig("user", "hwModel")).to eq("TBEAM")
+      expect(normalized.dig("user", "publicKey")).to eq("pk")
+      expect(normalized.dig("user", "isUnmessagable")).to be(true)
+      expect(normalized.dig("deviceMetrics", "batteryLevel")).to eq(87)
+      expect(normalized.dig("deviceMetrics", "channelUtilization")).to eq(12.5)
+      expect(normalized.dig("deviceMetrics", "airUtilTx")).to eq(2.75)
+      expect(normalized.dig("deviceMetrics", "uptimeSeconds")).to eq(1200)
+      expect(normalized.dig("position", "precisionBits")).to eq(13)
+      expect(normalized.dig("position", "locationSource")).to eq("LOC_MANUAL")
+      expect(normalized["lastHeard"]).to eq(reference_time.to_i)
+      expect(normalized["hopsAway"]).to eq(2)
+      expect(normalized["isFavorite"]).to be(true)
+      expect(normalized["hwModel"]).to eq("TBEAM")
+    end
+
+    it "rejects malformed normalized nodeinfo payloads" do
+      invalid_payload = {
+        "user" => { "shortName" => "NODE" },
+        "deviceMetrics" => "invalid",
+        "position" => "invalid",
+      }
+
+      valid = PotatoMesh::Application.send(:valid_decrypted_nodeinfo_payload?, invalid_payload)
+      normalized = PotatoMesh::Application.send(:normalize_decrypted_nodeinfo_payload, nil)
+
+      expect(valid).to be(false)
+      expect(normalized).to eq({})
+    end
+
     it "prefers decrypted message fields over encrypted ones" do
       encrypted_payload = Base64.strict_encode64("cipher".b)
       encrypted_message = {

web/spec/config_spec.rb

@@ -239,6 +239,30 @@ RSpec.describe PotatoMesh::Config do
     end
   end
 
+  describe ".remote_instance_request_timeout" do
+    it "returns the baked-in request timeout when unset" do
+      within_env("REMOTE_INSTANCE_REQUEST_TIMEOUT" => nil) do
+        expect(described_class.remote_instance_request_timeout).to eq(
+          PotatoMesh::Config::DEFAULT_REMOTE_INSTANCE_REQUEST_TIMEOUT,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("REMOTE_INSTANCE_REQUEST_TIMEOUT" => "19") do
+        expect(described_class.remote_instance_request_timeout).to eq(19)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("REMOTE_INSTANCE_REQUEST_TIMEOUT" => "0") do
+        expect(described_class.remote_instance_request_timeout).to eq(
+          PotatoMesh::Config::DEFAULT_REMOTE_INSTANCE_REQUEST_TIMEOUT,
+        )
+      end
+    end
+  end
+
   describe ".federation_max_instances_per_response" do
     it "returns the baked-in response limit when unset" do
       within_env("FEDERATION_MAX_INSTANCES_PER_RESPONSE" => nil) do
@@ -359,6 +383,54 @@ RSpec.describe PotatoMesh::Config do
     end
   end
 
+  describe ".federation_shutdown_timeout_seconds" do
+    it "returns the default shutdown timeout when unset" do
+      within_env("FEDERATION_SHUTDOWN_TIMEOUT" => nil) do
+        expect(described_class.federation_shutdown_timeout_seconds).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_SHUTDOWN_TIMEOUT_SECONDS,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_SHUTDOWN_TIMEOUT" => "9") do
+        expect(described_class.federation_shutdown_timeout_seconds).to eq(9)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("FEDERATION_SHUTDOWN_TIMEOUT" => "-1") do
+        expect(described_class.federation_shutdown_timeout_seconds).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_SHUTDOWN_TIMEOUT_SECONDS,
+        )
+      end
+    end
+  end
+
+  describe ".federation_crawl_cooldown_seconds" do
+    it "returns the default crawl cooldown when unset" do
+      within_env("FEDERATION_CRAWL_COOLDOWN" => nil) do
+        expect(described_class.federation_crawl_cooldown_seconds).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_CRAWL_COOLDOWN_SECONDS,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_CRAWL_COOLDOWN" => "17") do
+        expect(described_class.federation_crawl_cooldown_seconds).to eq(17)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("FEDERATION_CRAWL_COOLDOWN" => "0") do
+        expect(described_class.federation_crawl_cooldown_seconds).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_CRAWL_COOLDOWN_SECONDS,
+        )
+      end
+    end
+  end
+
   describe ".db_path" do
     it "returns the default path inside the data directory" do
       expect(described_class.db_path).to eq(described_class.default_db_path)

web/spec/database_spec.rb

@@ -184,6 +184,65 @@ RSpec.describe PotatoMesh::App::Database do
     expect(hop_columns).to include("trace_id", "hop_index", "node_id")
   end
 
+  it "creates positions and neighbors tables when absent" do
+    SQLite3::Database.new(PotatoMesh::Config.db_path) do |db|
+      db.execute("CREATE TABLE nodes(node_id TEXT)")
+      db.execute("CREATE TABLE messages(id INTEGER PRIMARY KEY)")
+      db.execute("CREATE TABLE telemetry(id INTEGER PRIMARY KEY, rx_time INTEGER, rx_iso TEXT)")
+    end
+
+    expect(column_names_for("positions")).to be_empty
+    expect(column_names_for("neighbors")).to be_empty
+
+    harness_class.ensure_schema_upgrades
+
+    positions_columns = column_names_for("positions")
+    expect(positions_columns).to include("id", "node_id", "rx_time", "ingestor")
+
+    neighbors_columns = column_names_for("neighbors")
+    expect(neighbors_columns).to include("node_id", "neighbor_id", "rx_time", "ingestor")
+  end
+
+  it "adds ingestor columns to legacy positions neighbors and traces tables" do
+    SQLite3::Database.new(PotatoMesh::Config.db_path) do |db|
+      db.execute("CREATE TABLE nodes(node_id TEXT)")
+      db.execute("CREATE TABLE messages(id INTEGER PRIMARY KEY)")
+      db.execute("CREATE TABLE telemetry(id INTEGER PRIMARY KEY, rx_time INTEGER, rx_iso TEXT)")
+      db.execute <<~SQL
+                   CREATE TABLE positions (
+                     id INTEGER PRIMARY KEY,
+                     rx_time INTEGER,
+                     rx_iso TEXT,
+                     node_id TEXT
+                   )
+                 SQL
+      db.execute <<~SQL
+                   CREATE TABLE neighbors (
+                     node_id TEXT,
+                     neighbor_id TEXT,
+                     rx_time INTEGER
+                   )
+                 SQL
+      db.execute <<~SQL
+                   CREATE TABLE traces (
+                     id INTEGER PRIMARY KEY,
+                     request_id INTEGER,
+                     src TEXT,
+                     dest TEXT,
+                     rx_time INTEGER,
+                     rx_iso TEXT
+                   )
+                 SQL
+      db.execute("CREATE TABLE trace_hops(trace_id INTEGER, hop_index INTEGER, node_id TEXT)")
+    end
+
+    harness_class.ensure_schema_upgrades
+
+    expect(column_names_for("positions")).to include("ingestor")
+    expect(column_names_for("neighbors")).to include("ingestor")
+    expect(column_names_for("traces")).to include("ingestor")
+  end
+
   it "adds the contact_link column to existing instances tables" do
     SQLite3::Database.new(PotatoMesh::Config.db_path) do |db|
       db.execute("CREATE TABLE nodes(node_id TEXT)")

web/spec/federation_spec.rb

@@ -23,6 +23,9 @@ require "uri"
 require "socket"
 
 RSpec.describe PotatoMesh::App::Federation do
+  NODES_API_PATH = "/api/nodes".freeze
+  HTTP_CONNECTION_DOUBLE = "Net::HTTPConnection".freeze
+
   subject(:federation_helpers) do
     Class.new do
       extend PotatoMesh::App::Federation
@@ -57,6 +60,8 @@ RSpec.describe PotatoMesh::App::Federation do
             :federation_thread,
             :initial_federation_thread,
             :federation_worker_pool,
+            :federation_shutdown_requested,
+            :federation_shutdown_hook_installed,
           ).new
         end
 
@@ -77,10 +82,12 @@ RSpec.describe PotatoMesh::App::Federation do
     federation_helpers.instance_variable_set(:@remote_instance_verify_callback, nil)
     federation_helpers.reset_debug_messages
     federation_helpers.reset_warn_messages
+    federation_helpers.clear_federation_crawl_state!
     federation_helpers.shutdown_federation_worker_pool!
   end
 
   after do
+    federation_helpers.clear_federation_crawl_state!
     federation_helpers.shutdown_federation_worker_pool!
   end
 
@@ -270,7 +277,7 @@ RSpec.describe PotatoMesh::App::Federation do
     let(:response_map) do
       mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] }
       attributes_list.each do |attributes|
-        mapping[[attributes[:domain], "/api/nodes"]] = [node_payload, :nodes]
+        mapping[[attributes[:domain], NODES_API_PATH]] = [node_payload, :nodes]
         mapping[[attributes[:domain], "/api/instances"]] = [[], :instances]
       end
       mapping
@@ -331,7 +338,7 @@ RSpec.describe PotatoMesh::App::Federation do
       mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] }
       attributes_list.each_with_index do |attributes, index|
         mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = [node_payload, :nodes]
-        mapping[[attributes[:domain], "/api/nodes"]] = [node_payload, :nodes]
+        mapping[[attributes[:domain], NODES_API_PATH]] = [node_payload, :nodes]
         mapping[[attributes[:domain], "/api/instances"]] = [[], :instances]
         allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(payload_entries[index]).and_return([attributes, "signature-#{index}", nil])
       end
@@ -352,6 +359,70 @@ RSpec.describe PotatoMesh::App::Federation do
         [attributes_list[1][:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"],
         [attributes_list[2][:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"],
       )
+      expect(captured_paths).to include(
+        [attributes_list[0][:domain], NODES_API_PATH],
+        [attributes_list[1][:domain], NODES_API_PATH],
+        [attributes_list[2][:domain], NODES_API_PATH],
+      )
+      expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(node_payload.length))
+    end
+
+    it "falls back to full node data when the recent window request fails" do
+      now = Time.at(1_700_000_000)
+      allow(Time).to receive(:now).and_return(now)
+      allow(PotatoMesh::Config).to receive(:remote_instance_max_node_age).and_return(900)
+      recent_cutoff = now.to_i - 900
+
+      mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] }
+      attributes_list.each_with_index do |attributes, index|
+        mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = [nil, ["no window"]]
+        mapping[[attributes[:domain], NODES_API_PATH]] = [node_payload, :nodes]
+        mapping[[attributes[:domain], "/api/instances"]] = [[], :instances]
+        allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(payload_entries[index]).and_return([attributes, "signature-#{index}", nil])
+      end
+
+      captured_paths = []
+      allow(federation_helpers).to receive(:fetch_instance_json) do |host, path|
+        captured_paths << [host, path]
+        mapping.fetch([host, path]) { [nil, []] }
+      end
+      allow(federation_helpers).to receive(:verify_instance_signature).and_return(true)
+      allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil])
+      allow(federation_helpers).to receive(:upsert_instance_record)
+
+      federation_helpers.ingest_known_instances_from!(db, seed_domain)
+
+      expect(captured_paths).to include(
+        [attributes_list[0][:domain], NODES_API_PATH],
+        [attributes_list[1][:domain], NODES_API_PATH],
+        [attributes_list[2][:domain], NODES_API_PATH],
+      )
+      expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(be_nil)
+    end
+
+    it "falls back to recent node window when full node data is unavailable" do
+      now = Time.at(1_700_000_000)
+      allow(Time).to receive(:now).and_return(now)
+      allow(PotatoMesh::Config).to receive(:remote_instance_max_node_age).and_return(900)
+      recent_cutoff = now.to_i - 900
+
+      mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] }
+      attributes_list.each_with_index do |attributes, index|
+        mapping[[attributes[:domain], "/api/nodes?since=#{recent_cutoff}&limit=1000"]] = [node_payload, :nodes]
+        mapping[[attributes[:domain], NODES_API_PATH]] = [nil, ["full data unavailable"]]
+        mapping[[attributes[:domain], "/api/instances"]] = [[], :instances]
+        allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(payload_entries[index]).and_return([attributes, "signature-#{index}", nil])
+      end
+
+      allow(federation_helpers).to receive(:fetch_instance_json) do |host, path|
+        mapping.fetch([host, path]) { [nil, []] }
+      end
+      allow(federation_helpers).to receive(:verify_instance_signature).and_return(true)
+      allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil])
+      allow(federation_helpers).to receive(:upsert_instance_record)
+
+      federation_helpers.ingest_known_instances_from!(db, seed_domain)
+
       expect(attributes_list.map { |attrs| attrs[:nodes_count] }).to all(eq(node_payload.length))
     end
   end
@@ -549,7 +620,7 @@ RSpec.describe PotatoMesh::App::Federation do
     end
 
     it "applies federation headers to instance fetch requests" do
-      connection = instance_double("Net::HTTPConnection")
+      connection = instance_double(HTTP_CONNECTION_DOUBLE)
       success_response = Net::HTTPOK.new("1.1", "200", "OK")
       allow(success_response).to receive(:body).and_return("{}")
       allow(success_response).to receive(:code).and_return("200")
@@ -571,13 +642,56 @@ RSpec.describe PotatoMesh::App::Federation do
       expect(captured_request["User-Agent"]).to eq(federation_helpers.send(:federation_user_agent_header))
       expect(captured_request["Content-Type"]).to be_nil
     end
+
+    it "wraps non-success HTTP responses" do
+      connection = instance_double(HTTP_CONNECTION_DOUBLE)
+      failure_response = Net::HTTPBadGateway.new("1.1", "502", "Bad Gateway")
+      allow(failure_response).to receive(:code).and_return("502")
+
+      allow(http_client).to receive(:start) do |&block|
+        block.call(connection)
+      end
+      allow(connection).to receive(:request).and_return(failure_response)
+
+      expect do
+        federation_helpers.send(:perform_instance_http_request, uri)
+      end.to raise_error(
+        PotatoMesh::App::InstanceFetchError,
+        a_string_including("unexpected response 502"),
+      )
+    end
+  end
+
+  describe ".federation_sleep_with_shutdown" do
+    it "returns false when shutdown is requested during sleep" do
+      allow(Kernel).to receive(:sleep)
+      call_count = 0
+      allow(federation_helpers).to receive(:federation_shutdown_requested?) do
+        call_count += 1
+        call_count > 1
+      end
+
+      result = federation_helpers.federation_sleep_with_shutdown(1.0)
+
+      expect(result).to be(false)
+      expect(Kernel).to have_received(:sleep).at_least(:once)
+    end
+
+    it "returns true when the full delay elapses without shutdown" do
+      allow(Kernel).to receive(:sleep)
+      allow(federation_helpers).to receive(:federation_shutdown_requested?).and_return(false)
+
+      result = federation_helpers.federation_sleep_with_shutdown(0.01)
+
+      expect(result).to be(true)
+    end
   end
 
   describe ".announce_instance_to_domain" do
     let(:payload) { "{}" }
     let(:https_uri) { URI.parse("https://remote.mesh/api/instances") }
     let(:http_uri) { URI.parse("http://remote.mesh/api/instances") }
-    let(:http_connection) { instance_double("Net::HTTPConnection") }
+    let(:http_connection) { instance_double(HTTP_CONNECTION_DOUBLE) }
     let(:success_response) { Net::HTTPOK.new("1.1", "200", "OK") }
 
     before do
@@ -655,6 +769,14 @@ RSpec.describe PotatoMesh::App::Federation do
       expect(federation_helpers.ensure_federation_worker_pool!).to be_nil
     end
 
+    it "returns nil when federation shutdown has been requested" do
+      allow(federation_helpers).to receive(:federation_enabled?).and_return(true)
+      federation_helpers.request_federation_shutdown!
+
+      expect(federation_helpers.ensure_federation_worker_pool!).to be_nil
+      expect(federation_helpers.send(:settings).federation_worker_pool).to be_nil
+    end
+
     it "creates and memoizes the worker pool" do
       allow(federation_helpers).to receive(:federation_enabled?).and_return(true)
 
@@ -667,6 +789,69 @@ RSpec.describe PotatoMesh::App::Federation do
     end
   end
 
+  describe ".ensure_federation_shutdown_hook!" do
+    it "registers a single at_exit hook when called repeatedly" do
+      allow(federation_helpers).to receive(:at_exit)
+
+      federation_helpers.ensure_federation_shutdown_hook!
+      federation_helpers.ensure_federation_shutdown_hook!
+
+      expect(federation_helpers).to have_received(:at_exit).once
+      expect(federation_helpers.send(:settings).federation_shutdown_hook_installed).to be(true)
+    end
+
+    it "delegates hook installation from instances to the application class" do
+      class_with_instance = Class.new do
+        include PotatoMesh::App::Federation
+      end
+
+      expect(class_with_instance).to receive(:ensure_federation_shutdown_hook!).once
+      class_with_instance.new.ensure_federation_shutdown_hook!
+    end
+
+    it "uses ivar guard when hook-installed setting is unavailable" do
+      helper_without_hook_setting = Class.new do
+        extend PotatoMesh::App::Federation
+
+        class << self
+          def settings
+            @settings ||= Struct.new(:federation_thread, :initial_federation_thread, :federation_worker_pool, :federation_shutdown_requested).new
+          end
+
+          # No-op in this helper because tests only assert hook registration behavior.
+          def shutdown_federation_background_work!(timeout: nil); end
+        end
+      end
+
+      allow(helper_without_hook_setting).to receive(:at_exit)
+      helper_without_hook_setting.ensure_federation_shutdown_hook!
+      helper_without_hook_setting.ensure_federation_shutdown_hook!
+
+      expect(helper_without_hook_setting).to have_received(:at_exit).once
+      expect(
+        helper_without_hook_setting.instance_variable_get(:@federation_shutdown_hook_installed),
+      ).to be(true)
+    end
+  end
+
+  describe ".stop_federation_thread!" do
+    it "wakes, joins, and kills a stubborn live thread" do
+      thread = instance_double(Thread)
+      allow(thread).to receive(:alive?).and_return(true, true, false)
+      allow(thread).to receive(:respond_to?).with(:wakeup).and_return(true)
+      allow(thread).to receive(:wakeup).and_raise(ThreadError, "not asleep")
+      allow(thread).to receive(:join)
+      allow(thread).to receive(:kill)
+
+      federation_helpers.set(:federation_thread, thread)
+      federation_helpers.stop_federation_thread!(:federation_thread, timeout: 0.01)
+
+      expect(thread).to have_received(:join).with(0.01)
+      expect(thread).to have_received(:kill)
+      expect(federation_helpers.send(:settings).federation_thread).to be_nil
+    end
+  end
+
   describe ".shutdown_federation_worker_pool!" do
     it "logs an error when shutdown fails" do
       pool = instance_double(PotatoMesh::App::WorkerPool)
@@ -683,6 +868,10 @@ RSpec.describe PotatoMesh::App::Federation do
   describe ".enqueue_federation_crawl" do
     let(:pool) { instance_double(PotatoMesh::App::WorkerPool) }
 
+    before do
+      allow(PotatoMesh::Config).to receive(:federation_crawl_cooldown_seconds).and_return(300)
+    end
+
     it "returns false and logs when the pool is unavailable" do
       allow(federation_helpers).to receive(:federation_worker_pool).and_return(nil)
 
@@ -696,6 +885,17 @@ RSpec.describe PotatoMesh::App::Federation do
       expect(federation_helpers.debug_messages.last).to include("Skipped remote instance crawl")
     end
 
+    it "returns false and logs when the domain is invalid" do
+      result = federation_helpers.enqueue_federation_crawl(
+        "https://bad domain",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+
+      expect(result).to be(false)
+      expect(federation_helpers.warn_messages.last).to include("Skipped remote instance crawl")
+    end
+
     it "schedules ingestion work on the pool" do
       allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
       db = instance_double(SQLite3::Database)
@@ -746,6 +946,29 @@ RSpec.describe PotatoMesh::App::Federation do
       expect(result).to be(false)
     end
 
+    it "does not apply cooldown when scheduling fails due to queue saturation" do
+      allow(PotatoMesh::Config).to receive(:federation_crawl_cooldown_seconds).and_return(300)
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      allow(pool).to receive(:schedule).and_raise(PotatoMesh::App::WorkerPool::QueueFullError, "full")
+
+      first = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 1,
+        overall_limit: 2,
+      )
+      second = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 1,
+        overall_limit: 2,
+      )
+
+      expect(first).to be(false)
+      expect(second).to be(false)
+      expect(federation_helpers.debug_messages).not_to include(
+        a_string_including("recent crawl completed"),
+      )
+    end
+
     it "logs when the worker pool is shutting down" do
       allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
       allow(pool).to receive(:schedule).and_raise(PotatoMesh::App::WorkerPool::ShutdownError, "closed")
@@ -766,6 +989,224 @@ RSpec.describe PotatoMesh::App::Federation do
 
       expect(result).to be(false)
     end
+
+    it "deduplicates crawls while a domain crawl is already in flight" do
+      db = instance_double(SQLite3::Database)
+      allow(db).to receive(:close)
+      captured_job = nil
+
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      allow(pool).to receive(:schedule) do |&block|
+        captured_job = block
+        instance_double(PotatoMesh::App::WorkerPool::Task)
+      end
+      allow(federation_helpers).to receive(:open_database).and_return(db)
+      allow(federation_helpers).to receive(:ingest_known_instances_from!)
+
+      first = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+      second = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+
+      expect(first).to be(true)
+      expect(second).to be(false)
+      expect(captured_job).not_to be_nil
+      captured_job.call
+      expect(db).to have_received(:close)
+    end
+
+    it "releases the crawl slot when opening the database fails" do
+      allow(federation_helpers).to receive(:federation_crawl_cooldown_seconds).and_return(0)
+      captured_job = nil
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      allow(pool).to receive(:schedule) do |&block|
+        captured_job = block
+        instance_double(PotatoMesh::App::WorkerPool::Task)
+      end
+      allow(federation_helpers).to receive(:open_database).and_raise(SQLite3::Exception, "db unavailable")
+      allow(federation_helpers).to receive(:ingest_known_instances_from!)
+
+      first = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+      expect(first).to be(true)
+      expect(captured_job).not_to be_nil
+
+      expect { captured_job.call }.to raise_error(SQLite3::Exception, "db unavailable")
+
+      second = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+      expect(second).to be(true)
+    end
+
+    it "deduplicates crawls across instance receivers using shared class state" do
+      helper_class = Class.new do
+        include PotatoMesh::App::Federation
+
+        class << self
+          attr_accessor :pool
+
+          def settings
+            @settings ||= Struct.new(:federation_shutdown_requested).new(false)
+          end
+
+          def set(key, value)
+            settings.public_send("#{key}=", value)
+          end
+
+          def federation_worker_pool
+            pool
+          end
+
+          # No-op to keep the test helper minimal while satisfying federation logging calls.
+          def debug_log(*); end
+
+          # No-op to keep the test helper minimal while satisfying federation logging calls.
+          def warn_log(*); end
+        end
+
+        def settings
+          self.class.settings
+        end
+
+        def set(key, value)
+          self.class.set(key, value)
+        end
+
+        def debug_log(...)
+          self.class.debug_log(...)
+        end
+
+        def warn_log(...)
+          self.class.warn_log(...)
+        end
+      end
+
+      pool_double = instance_double(PotatoMesh::App::WorkerPool)
+      allow(pool_double).to receive(:schedule).and_return(instance_double(PotatoMesh::App::WorkerPool::Task))
+      helper_class.pool = pool_double
+
+      first_receiver = helper_class.new
+      second_receiver = helper_class.new
+
+      first = first_receiver.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 1,
+        overall_limit: 2,
+      )
+      second = second_receiver.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 1,
+        overall_limit: 2,
+      )
+
+      expect(first).to be(true)
+      expect(second).to be(false)
+      expect(pool_double).to have_received(:schedule).once
+    end
+  end
+
+  describe ".fetch_instance_json" do
+    it "short-circuits when shutdown has been requested" do
+      federation_helpers.request_federation_shutdown!
+
+      payload, metadata = federation_helpers.fetch_instance_json("remote.mesh", NODES_API_PATH)
+
+      expect(payload).to be_nil
+      expect(metadata).to eq(["federation shutdown requested"])
+    end
+
+    it "stops iterating URI candidates after shutdown is requested mid-loop" do
+      calls = 0
+      allow(federation_helpers).to receive(:instance_uri_candidates).and_return([
+        URI.parse("https://remote.mesh/api/nodes"),
+        URI.parse("http://remote.mesh/api/nodes"),
+      ])
+      allow(federation_helpers).to receive(:perform_instance_http_request) do |_uri|
+        calls += 1
+        federation_helpers.request_federation_shutdown!
+        raise PotatoMesh::App::InstanceFetchError, "boom"
+      end
+
+      payload, metadata = federation_helpers.fetch_instance_json("remote.mesh", NODES_API_PATH)
+
+      expect(payload).to be_nil
+      expect(calls).to eq(1)
+      expect(metadata.first).to include("boom")
+    end
+  end
+
+  describe ".claim_federation_crawl_slot" do
+    it "initializes crawl dedupe state safely under concurrent access" do
+      federation_helpers.instance_variable_set(:@federation_crawl_mutex, nil)
+      federation_helpers.instance_variable_set(:@federation_crawl_in_flight, nil)
+      federation_helpers.instance_variable_set(:@federation_crawl_last_completed_at, nil)
+      federation_helpers.instance_variable_set(:@federation_crawl_init_mutex, nil)
+
+      threads = Array.new(12) do
+        Thread.new do
+          federation_helpers.initialize_federation_crawl_state!
+        end
+      end
+      threads.each(&:join)
+
+      mutex = federation_helpers.instance_variable_get(:@federation_crawl_mutex)
+      in_flight = federation_helpers.instance_variable_get(:@federation_crawl_in_flight)
+      last_completed = federation_helpers.instance_variable_get(:@federation_crawl_last_completed_at)
+
+      expect(mutex).to be_a(Mutex)
+      expect(in_flight).to be_a(Set)
+      expect(last_completed).to be_a(Hash)
+      expect(in_flight).to be_empty
+      expect(last_completed).to be_empty
+    end
+
+    it "returns cooldown when the domain completed recently" do
+      allow(PotatoMesh::Config).to receive(:federation_crawl_cooldown_seconds).and_return(300)
+      federation_helpers.clear_federation_crawl_state!
+      federation_helpers.release_federation_crawl_slot("remote.mesh")
+
+      result = federation_helpers.claim_federation_crawl_slot("remote.mesh")
+
+      expect(result).to eq(:cooldown)
+    end
+  end
+
+  describe ".shutdown_federation_background_work!" do
+    it "marks shutdown and clears announcer references" do
+      initial_thread = instance_double(Thread)
+      recurring_thread = instance_double(Thread)
+      pool = instance_double(PotatoMesh::App::WorkerPool)
+      allow(PotatoMesh::Config).to receive(:federation_shutdown_timeout_seconds).and_return(0.05)
+      allow(PotatoMesh::Config).to receive(:federation_task_timeout_seconds).and_return(0.05)
+
+      [initial_thread, recurring_thread].each do |thread|
+        allow(thread).to receive(:alive?).and_return(false)
+      end
+      allow(pool).to receive(:shutdown)
+
+      federation_helpers.set(:initial_federation_thread, initial_thread)
+      federation_helpers.set(:federation_thread, recurring_thread)
+      federation_helpers.set(:federation_worker_pool, pool)
+
+      federation_helpers.shutdown_federation_background_work!
+
+      expect(federation_helpers.federation_shutdown_requested?).to be(true)
+      expect(federation_helpers.send(:settings).initial_federation_thread).to be_nil
+      expect(federation_helpers.send(:settings).federation_thread).to be_nil
+      expect(federation_helpers.send(:settings).federation_worker_pool).to be_nil
+    end
   end
 
   describe ".wait_for_federation_tasks" do
@@ -836,4 +1277,32 @@ RSpec.describe PotatoMesh::App::Federation do
       federation_helpers.announce_instance_to_all_domains
     end
   end
+
+  describe ".start_federation_announcer!" do
+    it "clears shutdown, installs hook, and exits loop when sleep aborts" do
+      thread_double = instance_double(Thread)
+      captured = nil
+
+      allow(federation_helpers).to receive(:federation_enabled?).and_return(true)
+      allow(federation_helpers).to receive(:clear_federation_shutdown_request!)
+      allow(federation_helpers).to receive(:ensure_federation_shutdown_hook!)
+      allow(federation_helpers).to receive(:federation_sleep_with_shutdown).and_return(false)
+      allow(Thread).to receive(:new) do |&block|
+        captured = block
+        thread_double
+      end
+      allow(thread_double).to receive(:respond_to?).with(:name=).and_return(false)
+      allow(thread_double).to receive(:respond_to?).with(:daemon=).and_return(false)
+      allow(federation_helpers).to receive(:set)
+
+      result = federation_helpers.start_federation_announcer!
+      expect(result).to eq(thread_double)
+      expect(captured).to be_a(Proc)
+      captured.call
+
+      expect(federation_helpers).to have_received(:clear_federation_shutdown_request!)
+      expect(federation_helpers).to have_received(:ensure_federation_shutdown_hook!)
+      expect(federation_helpers).to have_received(:federation_sleep_with_shutdown)
+    end
+  end
 end