Fix ingest indexing, stream replay and shutdown reliability

2026-10-22 21:51:46 +00:00 · 2026-10-02 18:07:23 -04:00
1 parent 81e36352d0
commit 69077137e8
@@ -1,6 +1,6 @@
 # Copy this file to .env and configure your MQTT feed. No map key or build is needed.
 # Keep .env and any optional .secrets files private.
-CARTOLITE_IMAGE=ghcr.io/n30nex/cartolite-server:0.6.2
+CARTOLITE_IMAGE=ghcr.io/n30nex/cartolite-server:0.6.3
 CARTOLITE_BIND_ADDR=127.0.0.1
 CARTOLITE_PORT=8080
 # Optional CARTO override: see compose.carto.yml and docs/deployment.md.
@@ -9,7 +9,8 @@ CARTOLITE_PORT=8080
 # Your standard MeshCore MQTT feed.
 MQTT_BROKER_URL=tcp://broker.example.invalid:1883
 MQTT_TOPIC=meshcore/#
-MQTT_CLIENT_ID=cartolite-server
+# Leave empty for a unique clean-session ID, or set an ID unique to this instance.
+MQTT_CLIENT_ID=
 MQTT_USERNAME=
 MQTT_PASSWORD=
 # Empty accepts every valid region. Set exact comma-separated region labels to filter.
@@ -35,6 +35,9 @@ jobs:
        run: go vet ./...
      - working-directory: backend
        run: go test -race ./...
+      - name: Synthetic ingest index benchmark
+        working-directory: backend
+        run: go test ./internal/engine -run '^$' -bench BenchmarkNodeRepresentative -benchmem -benchtime=200ms -count=3

  frontend:
    name: Frontend test and build
@@ -197,7 +200,8 @@ jobs:
          candidate="${IMAGE_NAME}:sha-${GITHUB_SHA}-${{ matrix.arch }}"
          docker tag cartolite-server:ci "$candidate"
          docker push "$candidate"
-          digest="$(docker buildx imagetools inspect "$candidate" | awk '$1 == "Digest:" {print $2; exit}')"
+          docker buildx imagetools inspect "$candidate" > "$RUNNER_TEMP/candidate-inspect.txt"
+          digest="$(awk '$1 == "Digest:" {print $2; exit}' "$RUNNER_TEMP/candidate-inspect.txt")"
          [[ "$digest" =~ ^sha256:[0-9a-f]{64}$ ]]
          mkdir -p artifacts/candidate
          printf '%s\n' "$digest" > artifacts/candidate/${{ matrix.arch }}.txt
@@ -269,7 +273,8 @@ jobs:
          candidate="${IMAGE_NAME}:sha-${GITHUB_SHA}"
          docker buildx imagetools create --tag "$candidate" "${IMAGE_NAME}@${amd64}" "${IMAGE_NAME}@${arm64}"
          docker buildx imagetools inspect "$candidate" --raw | jq -e '[.manifests[].platform | .os + "/" + .architecture] | sort == ["linux/amd64","linux/arm64"]'
-          digest="$(docker buildx imagetools inspect "$candidate" | awk '$1 == "Digest:" {print $2; exit}')"
+          docker buildx imagetools inspect "$candidate" > "$RUNNER_TEMP/candidate-inspect.txt"
+          digest="$(awk '$1 == "Digest:" {print $2; exit}' "$RUNNER_TEMP/candidate-inspect.txt")"
          [[ "$digest" =~ ^sha256:[0-9a-f]{64}$ ]]
          echo "digest=$digest" >> "$GITHUB_OUTPUT"
          echo "image=${IMAGE_NAME}@${digest}" >> "$GITHUB_OUTPUT"
@@ -1,5 +1,14 @@
 # Changelog

+## 0.6.3 - Reliability and ingest performance
+
+- Replace full-table node identity scans on every update with an incremental index; retain known positions across coordinate-free region aliases and prevent stale alias events from moving nodes backwards.
+- Keep reconnect catch-up in state and route counts without replaying it as fresh animation, sound, or Follow activity.
+- Avoid rebuilding attached route geometry when only a node label or freshness changes.
+- Drain accepted observations into the final checkpoint and cancel active event streams during shutdown.
+- Generate unique clean-session MQTT identities for unconfigured installations; explicit broker IDs remain unchanged. Clear an old shared default when upgrading multiple installs on one broker.
+- Record the end-to-end audit, synthetic regressions and index benchmark in docs/audit-2026-10-02.md.
+
 ## 0.6.2 - Compact map controls

 - Fit the map legend into the bottom-left corner.
@@ -1 +1 @@
-0.6.2
+0.6.3
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"net"
 	"net/http"
 	"os"
 	"os/signal"
@@ -77,7 +78,7 @@ func run() error {
 		state.Wait()
 		return err
 	}
-	server := &http.Server{Addr: cfg.HTTPAddr, Handler: api.Handler(), ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 75 * time.Second}
+	server := newHTTPServer(ctx, cfg.HTTPAddr, api.Handler())
 	errors := make(chan error, 1)
 	go func() {
 		log.Info("cartolite listening", "address", cfg.HTTPAddr, "broker", mqtt.RedactBroker(cfg.MQTTBrokerURL), "version", cfg.Version, "gitSha", cfg.GitSHA)
@@ -97,11 +98,17 @@ func run() error {
 	}
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	if err := server.Shutdown(shutdownCtx); err != nil {
-		return err
-	}
+	shutdownErr := server.Shutdown(shutdownCtx)
 	state.Wait()
-	return nil
+	return shutdownErr
+}
+
+func newHTTPServer(ctx context.Context, addr string, handler http.Handler) *http.Server {
+	return &http.Server{
+		Addr: addr, Handler: handler,
+		ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 75 * time.Second,
+		BaseContext: func(net.Listener) context.Context { return ctx },
+	}
 }

 func reportOperationalStats(ctx context.Context, log *slog.Logger, state *engine.Engine, hub *httpapi.Hub) {
@@ -0,0 +1,42 @@
+package main
+
+import (
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+)
+
+func TestShutdownCancelsAnActiveStream(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	server := newHTTPServer(ctx, "127.0.0.1:0", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, "event: hello\ndata: {}\n\n")
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	listener, err := net.Listen("tcp", server.Addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer server.Close()
+	go func() { _ = server.Serve(listener) }()
+	client := &http.Client{Timeout: 3 * time.Second}
+	response, err := client.Get("http://" + listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	cancel()
+	shutdownCtx, stop := context.WithTimeout(context.Background(), time.Second)
+	defer stop()
+	if err := server.Shutdown(shutdownCtx); err != nil {
+		t.Fatalf("active SSE request prevented shutdown: %v", err)
+	}
+	if _, err := io.ReadAll(response.Body); err != nil {
+		t.Fatal(err)
+	}
+}
@@ -0,0 +1,18 @@
+package config
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+)
+
+// Clean MQTT sessions do not need a stable identity across process restarts.
+// Keep generated IDs below the legacy 23-byte limit and avoid disconnecting
+// another default installation that uses the same broker.
+func defaultClientID() (string, error) {
+	var suffix [6]byte
+	if _, err := rand.Read(suffix[:]); err != nil {
+		return "", fmt.Errorf("generate MQTT client identity: %w", err)
+	}
+	return "cartolite-" + hex.EncodeToString(suffix[:]), nil
+}
@@ -0,0 +1,27 @@
+package config
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestClientIdentityDefaultsAreUniqueAndExplicitIDsArePreserved(t *testing.T) {
+	t.Setenv("MQTT_ENABLED", "false")
+	t.Setenv("MQTT_CLIENT_ID", "")
+	first, err := Load("test", "fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := Load("test", "fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.MQTTClientID == second.MQTTClientID || !strings.HasPrefix(first.MQTTClientID, "cartolite-") || len(first.MQTTClientID) > 23 {
+		t.Fatal("default installations share a client ID or exceed the legacy limit")
+	}
+	t.Setenv("MQTT_CLIENT_ID", "my-existing-instance")
+	explicit, err := Load("test", "fixture")
+	if err != nil || explicit.MQTTClientID != "my-existing-instance" {
+		t.Fatal("explicit broker identity was changed")
+	}
+}
@@ -33,7 +33,7 @@ func Load(version, gitSHA string) (Config, error) {
 		MQTTEnabled:   envBool("MQTT_ENABLED", true),
 		MQTTBrokerURL: strings.TrimSpace(os.Getenv("MQTT_BROKER_URL")),
 		MQTTTopic:     env("MQTT_TOPIC", "meshcore/#"),
-		MQTTClientID:  env("MQTT_CLIENT_ID", "cartolite-server"),
+		MQTTClientID:  strings.TrimSpace(os.Getenv("MQTT_CLIENT_ID")),
 		MQTTUsername:  strings.TrimSpace(os.Getenv("MQTT_USERNAME")),
 		MQTTPassword:  strings.TrimSpace(os.Getenv("MQTT_PASSWORD")),
 		QueueSize:     envInt("MQTT_INGEST_QUEUE_SIZE", 4096),
@@ -46,6 +46,12 @@ func Load(version, gitSHA string) (Config, error) {
 		return Config{}, err
 	}
 	c.Regions = regions
+	if c.MQTTClientID == "" {
+		c.MQTTClientID, err = defaultClientID()
+		if err != nil {
+			return Config{}, err
+		}
+	}
 	if c.MapProvider != "openfreemap" && c.MapProvider != "carto" {
 		return Config{}, fmt.Errorf("BASEMAP_PROVIDER must be openfreemap or carto")
 	}
@@ -14,6 +14,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 	"unicode"
@@ -57,6 +58,8 @@ type Engine struct {
 	bootID               string
 	log                  *slog.Logger
 	input                chan mqtt.Message
+	inputMu              sync.RWMutex
+	stopping             bool
 	feedWake             chan struct{}
 	desiredFeed          atomic.Bool
 	dropped              atomic.Int64
@@ -127,10 +130,7 @@ func New(options Options) (*Engine, error) {
 		e.lastCheckpointAt.Store(info.ModTime().UnixMilli())
 	}
 	for key, node := range e.nodes {
-		id := nodePublicID(node)
-		if current := e.nodeIDs[id]; current == nil || node.LastSeen > current.LastSeen {
-			e.nodeIDs[id] = node
-		}
+		e.selectNodeID(node)
 		e.indexNode(key, node)
 	}
 	prunedRoutes, prunedNodes := e.pruneDurableState(time.Now().UnixMilli())
@@ -144,6 +144,11 @@ func New(options Options) (*Engine, error) {
 func (e *Engine) SetPublisher(publish func(Event)) { e.publish = publish }

 func (e *Engine) Submit(message mqtt.Message) bool {
+	e.inputMu.RLock()
+	defer e.inputMu.RUnlock()
+	if e.stopping {
+		return false
+	}
 	select {
 	case e.input <- message:
 		return true
@@ -173,6 +178,17 @@ func (e *Engine) Run(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
+			// Stop admission before draining so every accepted observation is saved.
+			e.inputMu.Lock()
+			e.stopping = true
+			e.inputMu.Unlock()
+			for len(e.input) > 0 {
+				message := <-e.input
+				e.processed.Add(1)
+				if e.process(message) {
+					dirtyCheckpoint = true
+				}
+			}
 			_, _ = e.flushCheckpoint(time.Now(), dirtyCheckpoint)
 			e.updateSnapshot(time.Now())
 			return
@@ -309,8 +325,8 @@ func (e *Engine) observePublisher(message mqtt.Message) (*privateNode, bool) {
 		node.LastSeen = message.HeardAt
 		changed = true
 	}
-	e.refreshNodeID(nodePublicID(node))
-	if node.HasCoords && (topologyChanged || shouldPublishFreshness(node, message.HeardAt)) {
+	representativeChanged := e.selectNodeID(node)
+	if node.HasCoords && (representativeChanged || topologyChanged || shouldPublishFreshness(node, message.HeardAt)) {
 		e.emitNode(node)
 	}
 	return node, changed
@@ -409,7 +425,6 @@ func (e *Engine) upsertNode(region, key, name, role string, observer bool, lat,
 		node = &privateNode{Region: region, Key: key, Role: normalizeRole(role), Observer: observer, LastSeen: seenAt}
 		node.Label = sanitizeLabel(name, node.Role, observer)
 		e.nodes[mapKey] = node
-		e.nodeIDs[nodePublicID(node)] = node
 		e.indexNode(mapKey, node)
 	}
 	changed := created
@@ -440,14 +455,19 @@ func (e *Engine) upsertNode(region, key, name, role string, observer bool, lat,
 		node.LastSeen = seenAt
 		changed = true
 	}
-	e.refreshNodeID(nodePublicID(node))
-	if node.HasCoords && (topologyChanged || shouldPublishFreshness(node, seenAt)) {
+	representativeChanged := e.selectNodeID(node)
+	if node.HasCoords && (representativeChanged || topologyChanged || shouldPublishFreshness(node, seenAt)) {
 		e.emitNode(node)
 	}
 	return node, changed
 }

 func (e *Engine) emitNode(node *privateNode) {
+	// Region aliases share a public ID. Older aliases must not overwrite the
+	// representative used by snapshots with a stale location or label over SSE.
+	if e.nodeIDs[nodePublicID(node)] != node {
+		return
+	}
 	node.LastPublished = node.LastSeen
 	seq := e.seq.Add(1)
 	e.emit(Event{Name: "node", Seq: seq, Data: NodeEventV2{Seq: seq, Node: publicNode(node)}})
@@ -676,15 +696,25 @@ func (e *Engine) evictOldestNode() {
 	}
 }

+// Prefer a known position, then the most recent alias. LastSeen only increases,
+// so updates can maintain the representative in O(1).
+// Removal still uses the full scan to find the best remaining region alias.
+func (e *Engine) selectNodeID(node *privateNode) bool {
+	id := nodePublicID(node)
+	if current := e.nodeIDs[id]; current == nil || (node.HasCoords && !current.HasCoords) || (node.HasCoords == current.HasCoords && node.LastSeen > current.LastSeen) {
+		e.nodeIDs[id] = node
+		return true
+	}
+	return false
+}
+
 func (e *Engine) refreshNodeID(id string) {
 	delete(e.nodeIDs, id)
 	for _, node := range e.nodes {
 		if nodePublicID(node) != id {
 			continue
 		}
-		if current := e.nodeIDs[id]; current == nil || node.LastSeen > current.LastSeen {
-			e.nodeIDs[id] = node
-		}
+		e.selectNodeID(node)
 	}
 }

@@ -0,0 +1,130 @@
+package engine
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/n30nex/cartolite-server/backend/internal/meshcore"
+	"github.com/n30nex/cartolite-server/backend/internal/mqtt"
+)
+
+func TestRegionAliasesDoNotRegressPublicNodeEvents(t *testing.T) {
+	state := newTestEngine(t)
+	now := time.Now().UnixMilli()
+	older, _ := state.upsertNode("YKF", "AA112233", "Older", "repeater", false, 43.4, -80.4, true, now)
+	newer, _ := state.upsertNode("YYZ", "AA112233", "Newer", "repeater", false, 43.5, -80.5, true, now+10)
+	var events []NodeV2
+	state.SetPublisher(func(event Event) {
+		if event.Name == "node" {
+			events = append(events, event.Data.(NodeEventV2).Node)
+		}
+	})
+	state.upsertNode("YKF", "AA112233", "Old update", "repeater", false, 43.6, -80.6, true, now+2)
+	if state.nodeIDs[nodePublicID(older)] != newer || len(events) != 0 {
+		t.Fatal("older region alias replaced the current public node")
+	}
+	state.upsertNode("YKF", "AA112233", "Latest", "repeater", false, 43.7, -80.7, true, now+20)
+	if state.nodeIDs[nodePublicID(older)] != older || len(events) != 1 || events[0].Lat != 43.7 {
+		t.Fatal("newest observation did not become the public representative")
+	}
+	// A switch back within the freshness throttle must still publish the
+	// different representative, even if that alias's own metadata did not move.
+	state.upsertNode("YYZ", "AA112233", "Newer", "repeater", false, 43.5, -80.5, true, now+30)
+	if len(events) != 2 || events[1].Lat != 43.5 {
+		t.Fatal("freshness throttle hid a changed public representative")
+	}
+	state.upsertNode("YKF", "AA112233", "Latest", "repeater", false, 43.7, -80.7, true, now+40)
+	delete(state.nodes, nodeMapKey(older.Region, older.Key))
+	state.refreshNodeID(nodePublicID(older))
+	if state.nodeIDs[nodePublicID(older)] != newer {
+		t.Fatal("removal lost the surviving region alias")
+	}
+}
+
+func TestShutdownPersistsEveryAcceptedQueuedObservation(t *testing.T) {
+	state := newTestEngine(t)
+	const count = 20
+	for i := 1; i <= count; i++ {
+		if !state.Submit(mqtt.Message{
+			Topic:   mqtt.Topic{Region: "YKF", PublisherKey: fmt.Sprintf("%08X", i), Kind: "status"},
+			Payload: map[string]any{"latitude": 43.4, "longitude": -80.4, "name": "Fixture"},
+			HeardAt: time.Now().UnixMilli(),
+		}) {
+			t.Fatal("bounded fixture was not accepted")
+		}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	go state.Run(ctx)
+	select {
+	case <-state.done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("shutdown did not finish")
+	}
+	if state.Submit(mqtt.Message{}) || state.QueueDepth() != 0 || state.OperationalStats().Processed != count {
+		t.Fatal("shutdown lost queued input or admitted work after the checkpoint")
+	}
+	nodes, _, err := loadCheckpoint(state.checkpoint)
+	if err != nil || len(nodes) != count {
+		t.Fatalf("accepted observations missing from final checkpoint: nodes=%d error=%v", len(nodes), err)
+	}
+}
+
+func TestUnpositionedRegionAliasRetainsKnownPositionWithoutResolvingMissingHop(t *testing.T) {
+	state := newTestEngine(t)
+	now := time.Now().UnixMilli()
+	positioned, _ := state.upsertNode("YKF", "AA112233", "Known", "repeater", false, 43.4, -80.4, true, now)
+	state.upsertNode("YYZ", "AA112233", "No position", "repeater", false, 0, 0, false, now+100)
+	id := nodePublicID(positioned)
+	if state.nodeIDs[id] != positioned {
+		t.Fatal("a coordinate-free alias hid the known public position")
+	}
+	// The public position index must never supply private RF resolution evidence.
+	if state.nodes[nodeMapKey("YYZ", positioned.Key)].HasCoords {
+		t.Fatal("coordinates leaked into an unpositioned private region record")
+	}
+	observer, _ := state.upsertNode("YYZ", "CC112233", "Observer", "unknown", true, 43.5, -80.5, true, now+100)
+	rssi := -70.0
+	segments, changed := state.resolveAndRecord(
+		mqtt.Message{Topic: mqtt.Topic{Region: "YYZ"}, RSSI: &rssi, HeardAt: now + 100},
+		meshcore.Packet{HashSize: 1, Path: []string{"AA"}, PayloadType: meshcore.PayloadControl}, nil, observer, "Control",
+	)
+	if changed || len(segments) != 0 {
+		t.Fatal("public position index created a guessed private hop")
+	}
+	state.refreshNodeID(id)
+	if state.nodeIDs[id] != positioned {
+		t.Fatal("index reconstruction lost the known public position")
+	}
+}
+
+// Compare the retained removal scan with the new per-observation index update.
+// Both operate on the same synthetic identity table; setup is outside timing.
+func BenchmarkNodeRepresentative(b *testing.B) {
+	for _, count := range []int{100, maxNodes} {
+		for _, method := range []string{"scan", "indexed"} {
+			b.Run(fmt.Sprintf("%d/%s", count, method), func(b *testing.B) {
+				state := &Engine{nodes: make(map[string]*privateNode), nodeIDs: make(map[string]*privateNode)}
+				var changed *privateNode
+				for i := 1; i <= count; i++ {
+					changed = &privateNode{Key: fmt.Sprintf("%08X", i), Region: "YKF", LastSeen: 1}
+					state.nodes[nodeMapKey(changed.Region, changed.Key)] = changed
+					state.selectNodeID(changed)
+				}
+				id := nodePublicID(changed)
+				b.ReportAllocs()
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					changed.LastSeen++
+					if method == "scan" {
+						state.refreshNodeID(id)
+					} else {
+						state.selectNodeID(changed)
+					}
+				}
+			})
+		}
+	}
+}
@@ -2,7 +2,7 @@ name: cartolite-server

 services:
  cartolite:
-    image: ${CARTOLITE_IMAGE:-ghcr.io/n30nex/cartolite-server:0.6.2}
+    image: ${CARTOLITE_IMAGE:-ghcr.io/n30nex/cartolite-server:0.6.3}
    restart: unless-stopped
    ports:
      - "${CARTOLITE_BIND_ADDR:-127.0.0.1}:${CARTOLITE_PORT:-8080}:8080"
@@ -11,7 +11,7 @@ services:
      MQTT_ENABLED: "true"
      MQTT_BROKER_URL: ${MQTT_BROKER_URL:?Set MQTT_BROKER_URL}
      MQTT_TOPIC: ${MQTT_TOPIC:-meshcore/#}
-      MQTT_CLIENT_ID: ${MQTT_CLIENT_ID:-cartolite-server}
+      MQTT_CLIENT_ID: ${MQTT_CLIENT_ID:-}
      MQTT_USERNAME: ${MQTT_USERNAME:-}
      MQTT_PASSWORD: ${MQTT_PASSWORD:-}
      REGION_ALLOWLIST: ${REGION_ALLOWLIST:-}
@@ -0,0 +1,69 @@
+# End-to-end reliability and performance audit — 2 October 2026
+
+## Scope and evidence
+
+This review covers Canada and Worldwide from MQTT input through normalization, decoding, private resolution, public state, checkpointing, HTTP/SSE delivery, Map/Netgraph rendering, Follow/audio, preferences, packaging and release. Canada Labs, the Android shell and the live Canada service are also included.
+
+Baseline: Canada 0.15.2 (`f2b4bdff314094c22749819ddc6817d545aa0fa0`); Worldwide 0.6.2 with release-workflow maintenance at `81e36352d03c023487ccc03626bfa8459959f4f6`. The corrective releases are Canada 0.15.3 and Worldwide 0.6.3.
+
+Source inspection and read-only operational checks established the findings below. All automated builds, tests, browser evidence and measurements run in GitHub Actions with synthetic fixtures. The implementation is subject to its exact-commit PR/main checks; this document does not declare a deployment complete.
+
+## Findings fixed in this patch
+
+| ID | Priority | Finding and consequence | Correction and evidence |
+|---|---|---|---|
+| B01 | P1 | Each node observation rebuilt its public representative by scanning and hashing the entire private node table. At the 10,000-node cap this makes ingest work grow with every stored node. | Maintain the representative incrementally. The full scan remains on removal, where replacement actually needs discovery. A paired 100/10,000-node benchmark compares scan and indexed updates on identical synthetic tables, with setup excluded and three samples. |
+| B02 | P1 | Multiple region records share one public identity, but every alias could emit its own node event. An older region record could overwrite the snapshot's newer position in connected browsers; a newer record without coordinates could hide a known position. | Prefer coordinate-bearing records, then the freshest alias; emit only the current representative. A representative switch bypasses the freshness throttle. Tests cover stale/new alias events, missing coordinates, removal fallback and unchanged fail-closed private hop resolution. |
+| B03 | P2 | SSE reconnect replay updated state and was also dispatched as fresh packet effects, audio and Follow input. Reconnecting could create an artificial burst of apparently live activity. | Use the existing hello sequence as the atomic catch-up boundary. Apply replay to state/counts only; new packets still animate and sound. Test initial/native reconnect boundaries, Map/Netgraph with sound enabled, and Canada's Labs caption. No public field or endpoint is added. |
+| B04 | P2 | Every node freshness/name update invalidated all connected route geometry even when its position was unchanged. | Only invalidate attached geometry when latitude/longitude changes or a node first appears. Node labels, freshness and inspector updates still flow. A regression distinguishes renaming from endpoint movement. |
+| B05 | P1 | Cancellation could leave accepted messages in the ingest queue outside the final checkpoint. Long-lived SSE handlers could also hold HTTP shutdown open until its deadline; the error return could skip waiting for the state writer. | Close ingest admission, drain the finite accepted queue, then flush. Give HTTP requests the process cancellation context and always wait for state shutdown. Tests load the final checkpoint and terminate an active stream. This preserves accepted input, not messages never delivered by MQTT. |
+| B06 | P2 | Two default installations used the same MQTT client ID. A shared broker may repeatedly disconnect them. | Generate a short random clean-session ID when configuration is blank. Explicit IDs remain unchanged. Test uniqueness/length and overrides; update Compose and installation examples. Existing operators must clear a copied shared default or choose distinct explicit IDs. |
+| B07 | P2 | Image digest extraction stopped reading the producer early under pipefail, which could fail an otherwise valid candidate publication or promotion. Canada lacked the existing Worldwide controlled retry path. | Read inspection output completely before extracting the digest. Port the tag-based manual release retry to Canada while retaining version, main ancestry, successful exact-main CI and attestation checks. Promote the tested digest without rebuilding. |
+
+The index is private to the engine. It does not supply evidence to route inference: an unpositioned or ambiguous regional hop stays unresolved. Packet classification, audio timing, route meaning, public API v2 and checkpoint v1 are unchanged.
+
+## End-to-end review coverage
+
+| Area | Reviewed paths and retained constraints |
+|---|---|
+| MQTT and decoder | Topic/region validation, bounded envelopes, timestamps, coordinates, RF values, raw decoding, reconnect/subscription handling and clean-session identity. No credentials or payload content are made public. |
+| Engine and checkpoint | Prefix ambiguity, missing coordinates/RF, forwarder roles, distance gates, aliases, node/route caps, retention, batching, atomic persistence, queue admission and shutdown. Fixes B01/B02/B05; private resolver rules remain intact. |
+| HTTP and SSE | Sanitized schema, subscriber/replay bounds, atomic replay ordering, gap/boot recovery, cancellation, cache/CSP policies and health/readiness. Fix B03 uses the existing hello boundary. |
+| Map and terrain | Incremental sources, route geometry/projection invalidation, adaptive sampling, retained historical paths, terrain/style recovery and custom-effect lifecycle. B04 avoids work without hiding routes or reducing fixture size. |
+| Netgraph | Stable layout, link caches, selection, area focus, packet dispatch, fallback rendering and basemap-independent operation. Replay is state-only in both rendering paths. |
+| Everyday UI | Dock, sheets, inspector, responsive bounds, layer/preset settings, cross-tab appearance, text/motion settings and attribution. Existing UI regressions remain required. |
+| Follow and sound | Ten-second director, fixed area scope, manual interruption, waiting behavior, packet visibility rules and playback limits. Catch-up no longer enters the live director/audio path. All 30 voices remain. |
+| Canada Labs | Single feed/render owner, experiment changes, shared motion/quality, captions and audio. Replay updates stored state without generating live captions/effects. |
+| Canada Android | HTTPS-only navigation, external/deep links, saved state, renderer recovery, lifecycle and SSL-error handling. No native change or APK publication in this patch. Physical-device acceptance of the existing 1.1.0 draft remains outstanding. |
+| Delivery and operation | Non-root scratch/read-only images, volumes/limits, Actions-only builds, exact-image promotion, multiarchitecture Worldwide packaging, encrypted deployment backup and isolated restore requirements. |
+
+## Production baseline
+
+Read-only Canada checks on 2 October found version 0.15.2, zero container restarts since 27 September, ready MQTT/checkpoint/queue, zero dropped observations and an empty queue. One resource sample was approximately 73.5 MiB memory and 15.14% CPU; the checkpoint was approximately 3.7 MB. These are operational observations, **not** a controlled before/after performance comparison. No live traffic capture is included in this repository.
+
+No CartoLite systemd units were registered on the Canada droplet or the Pi during this audit. The documented recurring backup/independent watchdog setup is therefore not established by those hosts. Previous release backups are not evidence of a recurring recovery service. This patch does not provision a storage account or webhook.
+
+There is no project-operated Worldwide live target in this release scope. Worldwide acceptance uses clean and upgrade installations of the published AMD64/ARM64 candidates in Actions.
+
+## Remaining work, in priority order
+
+| ID | Priority | Next work and acceptance |
+|---|---|---|
+| R01 | P1 operations | Establish recurring encrypted off-host backups and independent outage/recovery monitoring using the existing deployment tools. Verify storage/notification ownership, retention and an isolated restore; prove notification transitions without interrupting unrelated services. |
+| R02 | P2 correctness | Define MQTT retained/old-timestamp semantics explicitly. The retained flag is not currently carried into the normalized message and out-of-range timestamps may become receipt time. Add fixtures that retain useful node adverts without presenting retained packet history as new activity. B03 fixes SSE replay only. |
+| R03 | P2 resilience | Budget ingest memory by accepted message shape/bytes as well as count. The envelope and queue are bounded separately, but their worst-case product can exceed the container budget. Add bounded oversized-shape/burst fixtures and reject before retention. No production OOM was observed. |
+| R04 | P2 Netgraph | Define a bounded relocation policy for moving nodes: metadata refresh can leave geographic grouping/layout established from earlier coordinates. Test cross-area movement without disrupting stable layout, selection or date-line behavior. |
+| R05 | P2 dependencies | Address Canada's medium development-tool alerts for Vitest/mocker through a separately tested toolchain update. Worldwide Dependabot alerts were disabled/inaccessible to this audit; enable verified repository alert visibility with the owner. Existing runtime npm audit and image vulnerability gates remain required. No production exploit is claimed. |
+| R06 | P2 lifecycle | Add MQTT reconnect/subscription cancellation fault injection and restart-under-load evidence. The new shutdown tests prove accepted-queue persistence and active-stream cancellation; they do not simulate every network timing race. |
+| R07 | P3 measured performance | Profile Canada's Labs metrics snapshots and graph/layout work before changing artistic behavior. Per-frame filtering/copying is a candidate for optimization, not a measured bottleneck yet. |
+| R08 | Release gate | Complete physical-device acceptance of the signed Canada Android 1.1.0 candidate against the existing certificate before publication. Keep the current public APK until that gate is met. |
+
+The older roadmap contains historical “Planned” rows for features subsequently delivered by the cinematic release. Use the cinematic delivery notes/changelog for those completed features and this audit for current corrective work; do not interpret the old table as a fresh runtime assessment.
+
+## Validation and release acceptance
+
+Required Actions checks retain all existing limits: Go test/vet/race; frontend types/unit/build and size gates; privacy/topology contracts; synthetic ingest/load and checkpoint checks; all themes/presets and terrain/fallback recovery; Map/Netgraph/Labs desktop and mobile journeys; the full 4,000-node/7,000-route fixture; container identity/vulnerability checks; Android API 26/36 where applicable; native AMD64/ARM64 clean/upgrade Worldwide installs.
+
+The new microbenchmark measures representative lookup/update only. Report medians and allocations from the Actions logs; do not turn its speedup into an end-to-end throughput claim.
+
+After reviewed passing merges, tag the exact successful main revisions and promote their tested images. Canada deployment additionally requires a fresh target/digest check, encrypted off-host backup, isolated restore, exact-image cutover, public Map/Netgraph/Labs and live SSE checks, advancing checkpoints, and at least five minutes with healthy readiness and no drops/restarts. Retain the prior digest for rollback. Release/deployment receipts and Actions links provide completion evidence separately from this source audit.
@@ -4,7 +4,7 @@

 Docker Engine and Compose v2 are required. Published images support `linux/amd64` and `linux/arm64`, including 64-bit Raspberry Pi hosts. No Node, Go, source build or map-provider account is required.

-In a new directory, download `compose.yml` and `default.env.example` from the latest release. Save the environment example as `.env`, then set your MQTT broker, topic and any required username/password. Choose a unique `MQTT_CLIENT_ID` for each instance.
+In a new directory, download `compose.yml` and `default.env.example` from the latest release. Save the environment example as `.env`, then set your MQTT broker, topic and any required username/password. Leave `MQTT_CLIENT_ID` empty for an automatically unique clean-session identity, or set a unique ID if your broker requires one. Existing configured IDs are preserved; installations using the old shared `cartolite-server` default should clear it or choose their own ID.

 ```sh
 docker compose pull
@@ -25,7 +25,7 @@ The map is at `http://localhost:8080/`; Netgraph is at `/netgraph/`. Default map
 | `CARTOLITE_PORT` | `8080` | Host HTTP port |
 | `MQTT_BROKER_URL` | required | Your MeshCore broker URL (`tcp`, `ssl`, `ws`, or `wss`) |
 | `MQTT_TOPIC` | `meshcore/#` | Subscription filter |
-| `MQTT_CLIENT_ID` | `cartolite-server` | Must be unique per instance |
+| `MQTT_CLIENT_ID` | generated per process | Optional explicit ID; must be unique per instance |
 | `MQTT_USERNAME` / `MQTT_PASSWORD` | empty | Configure both when authentication is required |
 | `REGION_ALLOWLIST` | empty | Accept all valid regions, or exact comma-separated labels |
 | `MQTT_INGEST_QUEUE_SIZE` | `4096` | Bounded ingest queue |
@@ -1,5 +1,7 @@
 # Audit roadmap

+Latest reliability review: [2 October 2026 end-to-end audit](audit-2026-10-02.md). The paired 0.15.3 / 0.6.3 patch addresses ingest indexing, region aliases, replay, redraws, shutdown and installation identity. Exact-commit Actions and release receipts determine delivery status; older rows below are historical.
+
 Latest UI review: [2026-09-13 audit and refinement](ui-ux-audit-2026-09-13.md). The paired refinement implements A07, A08, A10 and part of A09, with layer organization and scoped reset; release evidence is separate.

 Updated: 12 September 2026. This roadmap turns the end-to-end audit into tracked delivery work. Audit IDs A01–A12 are retained; F IDs track features and R IDs track validation risks.
@@ -0,0 +1,26 @@
+import { expect, test } from '@playwright/test';
+import { visualFixture, visualPacket } from './visualFixtures';
+
+for (const path of ['/', '/netgraph/']) {
+  test(`reconnect catch-up stays silent and does not animate on ${path}`, async ({page}) => {
+    await visualFixture(page);
+    await page.goto(path);
+    await expect(page.locator('html')).toHaveAttribute('data-fixture-stream', 'ready');
+    await expect(page.locator(path === '/' ? '#map' : '#netgraph-stage')).toHaveAttribute('data-render-state', 'idle');
+    await page.locator('#sound-button').click();
+    await page.locator('#sound-toggle').click();
+    await expect(page.locator('#sound-state')).toHaveText('On');
+    await page.keyboard.press('Escape');
+    const renderer = page.locator(path === '/' ? '#packet-canvas' : '#netgraph-stage');
+    const before = await renderer.getAttribute('data-last-packet-kind');
+    const scheduledBefore = Number(await page.locator('#sound-activity').getAttribute('data-scheduled'));
+    await page.evaluate(() => window.dispatchEvent(new CustomEvent('visual-hello', {detail:{bootId:'visual-fixture',seq:2}})));
+    await visualPacket(page, 1, 'Trace');
+    await visualPacket(page, 2, 'Advert');
+    expect(await renderer.getAttribute('data-last-packet-kind')).toBe(before);
+    expect(Number(await page.locator('#sound-activity').getAttribute('data-scheduled'))).toBe(scheduledBefore);
+    await visualPacket(page, 3, 'Text');
+    await expect(renderer).toHaveAttribute('data-last-packet-kind', 'Text');
+    await expect.poll(() => page.locator('#sound-activity').getAttribute('data-scheduled').then(Number)).toBeGreaterThan(scheduledBefore);
+  });
+}
@@ -33,11 +33,12 @@ export async function visualFixture(page: Page, zoom = 10.2): Promise<void> {
      readyState = 1;
      onopen: ((event: Event) => void) | null = null;
      private receive = (event: Event) => this.dispatchEvent(new MessageEvent('packet', { data: JSON.stringify((event as CustomEvent).detail) }));
+      private hello = (event: Event) => this.dispatchEvent(new MessageEvent('hello', { data: JSON.stringify((event as CustomEvent).detail) }));
      constructor() {
-        super(); window.addEventListener('visual-packet', this.receive);
+        super(); window.addEventListener('visual-packet', this.receive); window.addEventListener('visual-hello', this.hello);
        setTimeout(() => { this.onopen?.(new Event('open')); document.documentElement.dataset.fixtureStream = 'ready'; }, 0);
      }
-      close(): void { this.readyState = 2; window.removeEventListener('visual-packet', this.receive); }
+      close(): void { this.readyState = 2; window.removeEventListener('visual-packet', this.receive); window.removeEventListener('visual-hello', this.hello); }
    }
    Object.defineProperty(window, 'EventSource', { value: FixtureStream });
  }, { center: visualCenter, zoom });
@@ -1,12 +1,12 @@
 {
  "name": "cartolite-server-web",
-  "version": "0.6.2",
+  "version": "0.6.3",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "cartolite-server-web",
-      "version": "0.6.2",
+      "version": "0.6.3",
      "dependencies": {
        "maplibre-gl": "6.4.1"
      },
@@ -1,6 +1,6 @@
 {
  "name": "cartolite-server-web",
-  "version": "0.6.2",
+  "version": "0.6.3",
  "private": true,
  "type": "module",
  "scripts": {
@@ -8,7 +8,7 @@ const initial: StateV2 = {
  seq: 7,
  serverTime: 1,
  status: { feed: 'connected', activity: 'active', dropped: 0, version: '0.4.0', gitSha: 'abc' },
-  map: { center: [0, 20], zoom: 1.4 },
+  map: { center: [-96, 56], zoom: 3.4 },
  nodes: [],
  routes: []
 };
@@ -79,7 +79,7 @@ describe('LiveFeed recovery', () => {

    expect(recover).toHaveBeenCalledTimes(1);
    expect(onPacket).toHaveBeenCalledTimes(6);
-    expect(onPacket).toHaveBeenCalledWith(expect.objectContaining({ seq: 26 }));
+    expect(onPacket).toHaveBeenCalledWith(expect.objectContaining({ seq: 26 }), false);
    feed.stop();
  });

@@ -105,6 +105,22 @@ describe('LiveFeed recovery', () => {
    feed.stop();
  });

+  it('marks catch-up packets for state-only application on initial connection and native reconnect', () => {
+    const onPacket = vi.fn();
+    const feed = new LiveFeed(initial, handlers({ onPacket }));
+    feed.start();
+    const source = MockEventSource.instances[0]!;
+    source.send('hello', { bootId: initial.bootId, seq: 9 });
+    for (const seq of [8, 9, 10]) source.send('packet', observerPacket(seq));
+    expect(onPacket.mock.calls.map(call => [call[0].seq, call[1]])).toEqual([[8, true], [9, true], [10, false]]);
+    source.onerror?.(new Event('error'));
+    source.onopen?.(new Event('open'));
+    source.send('hello', { bootId: initial.bootId, seq: 12 });
+    for (const seq of [11, 12, 13]) source.send('packet', observerPacket(seq));
+    expect(onPacket.mock.calls.slice(3).map(call => [call[0].seq, call[1]])).toEqual([[11, true], [12, true], [13, false]]);
+    feed.stop();
+  });
+
  it('recovers when the stream hello cannot establish a trustworthy boot cursor', async () => {
    const recover = vi.fn(async () => ({ ...initial, seq: 8 }));
    const onError = vi.fn();
@@ -34,7 +34,7 @@ export async function fetchState(signal?: AbortSignal): Promise<StateV2> {
 export interface LiveFeedHandlers {
  onConnection(connected: boolean): void;
  onNode(event: NodeEventV2): void;
-  onPacket(event: PacketEventV2): void;
+  onPacket(event: PacketEventV2, replayed: boolean): void;
  onStatus(event: StatusEventV2): void;
  recover(): Promise<StateV2>;
  onError(error: Error): void;
@@ -44,6 +44,7 @@ export class LiveFeed {
  private source?: EventSource;
  private bootId: string;
  private seq: number;
+  private replayThrough = 0;
  private recovering?: Promise<void>;
  private recoveryTimer?: number;
  private recoveryFailures = 0;
@@ -72,6 +73,7 @@ export class LiveFeed {

  private connect(): void {
    if (this.source || this.stopped) return;
+    this.replayThrough = this.seq;
    const cursor = new URLSearchParams({ bootId: this.bootId, after: String(this.seq) });
    const source = new EventSource(`/api/events?${cursor.toString()}`, { withCredentials: true });
    this.source = source;
@@ -96,7 +98,7 @@ export class LiveFeed {
      if (current()) this.handleSequenced<NodeEventV2>(event, this.handlers.onNode);
    });
    source.addEventListener('packet', (event) => {
-      if (current()) this.handleSequenced<PacketEventV2>(event, this.handlers.onPacket);
+      if (current()) this.handleSequenced<PacketEventV2>(event, packet => this.handlers.onPacket(packet, packet.seq <= this.replayThrough));
    });
    source.addEventListener('status', (event) => {
      if (current()) this.handleSequenced<StatusEventV2>(event, this.handlers.onStatus);
@@ -114,8 +116,14 @@ export class LiveFeed {
  private handleHello(raw: Event): void {
    try {
      const hello = parseEvent<HelloV2>(raw);
+      if (!Number.isSafeInteger(hello.seq) || hello.seq < 0 || typeof hello.bootId !== 'string' || !hello.bootId) {
+        throw new Error('invalid stream hello');
+      }
      if (hello.bootId !== this.bootId) {
        void this.requestRecovery();
+      } else {
+        // A hello precedes each initial connection and native reconnect replay.
+        this.replayThrough = hello.seq;
      }
    } catch (error) {
      this.report(error);
@@ -639,10 +639,10 @@ async function start(): Promise<void> {
      onNode(event) {
        liveStore.upsertNode(event.node, event.seq);
      },
-      onPacket(event) {
+      onPacket(event, replayed) {
        const packet = liveStore.applyPacket(event);
        lastUpdate.textContent = formatUpdate(event.at);
-        if (!packet) return;
+        if (!packet || replayed) return;
        liveAnimator.add(packet, { longHaul: potentialLongHaulPacket(packet) });
        const scheduled = uiPreferences.livePackets ? routeSonifier.play(packet) : 0;
        if (scheduled > 0) pulseSoundChrome(scheduled);
@@ -207,9 +207,9 @@ async function start(): Promise<void> {
      onNode(event) {
        liveStore.upsertNode(event.node, event.seq);
      },
-      onPacket(event) {
+      onPacket(event, replayed) {
        const packet = liveStore.applyPacket(event);
-        if (!packet) return;
+        if (!packet || replayed) return;
        graph.preparePacket(packet);
        graph.addPacket(packet);
        const noteCount = routeSonifier.play(packet);
@@ -13,7 +13,7 @@ const initial: StateV2 = {
  seq: 7,
  serverTime: 1,
  status: { feed: 'connected', activity: 'active', dropped: 0, version: '0.4.0', gitSha: 'abc' },
-  map: { center: [0, 20], zoom: 1.4 },
+  map: { center: [-96, 56], zoom: 3.4 },
  nodes,
  routes: []
 };
@@ -28,6 +28,17 @@ describe('sequenceAction', () => {
 });

 describe('LiveStore', () => {
+  it('keeps route geometry stable for freshness and label updates, but invalidates a moved endpoint', () => {
+    const store = new LiveStore({ ...initial, routes: [existingRoute()] });
+    const updates = vi.fn();
+    store.subscribe(updates);
+    store.upsertNode({ ...nodes[0]!, label: 'Renamed', lastSeen: 200 }, 8);
+    expect(updates).toHaveBeenLastCalledWith(expect.anything(), expect.objectContaining({ routeGeometry: [] }));
+    store.upsertNode({ ...nodes[0]!, lng: nodes[0]!.lng + 0.1, lastSeen: 300 }, 9);
+    expect(updates).toHaveBeenLastCalledWith(expect.anything(), expect.objectContaining({ routeGeometry: ['r1'] }));
+    store.destroy();
+  });
+
  it('upserts nodes without duplicating IDs and marks their routes dirty', () => {
    const route = existingRoute();
    const store = new LiveStore({ ...initial, routes: [route] });
@@ -82,6 +82,8 @@ export class LiveStore {

  upsertNode(node: NodeV2, seq: number): void {
    const index = this.nodeIndexes.get(node.id);
+    const previous = index === undefined ? undefined : this.current.nodes[index];
+    const moved = !previous || previous.lat !== node.lat || previous.lng !== node.lng;
    const nodes = [...this.current.nodes];
    if (index !== undefined) nodes[index] = node;
    else {
@@ -89,7 +91,7 @@ export class LiveStore {
      nodes.push(node);
    }
    this.current = { ...this.current, seq, nodes };
-    this.emit({ nodes: [node], routeGeometry: [...(this.nodeRoutes.get(node.id) ?? [])] });
+    this.emit({ nodes: [node], routeGeometry: moved ? [...(this.nodeRoutes.get(node.id) ?? [])] : [] });
  }

  updateStatus(status: StatusV2, seq: number): void {
@@ -1 +1 @@
 .6.2
 .6.3