feat: Data tab — packets placed by where they started, for any areas

.github/workflows/deploy.yml

@@ -217,6 +217,12 @@ jobs:
           go test ./...
           echo "--- Decrypt CLI tests passed ---"
 
+      - name: Test path watch module
+        run: |
+          set -e -o pipefail
+          cd internal/pathwatch && go test -race ./...
+          echo "--- pathwatch tests passed ---"
+
       - name: Test user-management modules (users, mailer)
         run: |
           set -e -o pipefail
@@ -575,6 +581,8 @@ jobs:
           suite 2 test-filter-ux-e2e.js BASE_URL=$B
           suite 1 test-channel-issue-1087-e2e.js BASE_URL=$B
           suite 1 test-channel-issue-1111-e2e.js BASE_URL=$B
+          suite 2 test-map-areas-overlay-e2e.js BASE_URL=$B
+          suite 2 test-packets-origin-e2e.js BASE_URL=$B
           suite 2 test-map-modal-fluid-e2e.js BASE_URL=$B
           suite 2 test-map-nodes-pagination-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B
           suite 3 test-observer-iata-1188-e2e.js BASE_URL=$B
@@ -688,6 +696,9 @@ jobs:
           suite 3 test-packets-scope-column.js CHROMIUM_REQUIRE=1 BASE_URL=$B
           suite 3 test-path-inspector-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B
           suite 3 test-pr-1490-live-map-gpu-animations-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B
+          suite 1 test-region-quick-picks.js
+          suite 3 test-data-tab.js
+          suite 3 test-data-watch.js
           suite 3 test-table-sort.js
           suite 1 test-touch-targets.js CHROMIUM_REQUIRE=1 BASE_URL=$B
           suite 3 test-live-dedup.js BASE_URL=$B

Dockerfile

@@ -56,6 +56,10 @@ ENV CC=zigcc CGO_ENABLED=1
 # Build server
 WORKDIR /build/server
 COPY cmd/server/go.mod cmd/server/go.sum ./
+COPY internal/areamap/ ../../internal/areamap/
+COPY internal/origin/ ../../internal/origin/
+COPY internal/dataindex/ ../../internal/dataindex/
+COPY internal/pathwatch/ ../../internal/pathwatch/
 COPY internal/geofilter/ ../../internal/geofilter/
 COPY internal/sigvalidate/ ../../internal/sigvalidate/
 COPY internal/packetpath/ ../../internal/packetpath/

MIGRATIONS.md

@@ -29,9 +29,12 @@ if err := s.RunAsyncMigration(ctx, "my_migration_v1",
   table **immediately** — the ingestor boots and starts ingesting.
 - `fn` runs in a goroutine; the WaitGroup is shared with the rest of the
   ingestor (`Store.WaitForAsyncMigrations()` waits for everything).
-- On success the row flips to `done`; on error/panic to `failed` with the
-  error message captured.
-- Idempotent: rows in `done` state short-circuit; `failed`/`pending_async`
+- On success the row flips to `done`. A returned error matching
+  `context.Canceled` (including a wrapped error) becomes `cancelled`, such
+  as when shutdown interrupts a backfill. Other errors, including deadline
+  expiry, and panics become `failed` even if the context was cancelled.
+  Both interrupted states retain the error message and completion time.
+- Idempotent: rows in `done` state short-circuit; `failed`/`cancelled`/`pending_async`
   rows are retried on the next boot.
 
 Reference implementations: `Store.BackfillPathJSONAsync` (path_json
@@ -91,7 +94,7 @@ What this means for async migrations:
   running against the same DB is not a supported deployment shape.
 - **Within a single process**, concurrent `RunAsyncMigration(name=X)`
   callers race the initial `SELECT status` → `UPDATE/INSERT` step. The
-  current implementation re-schedules `fn` on a pending/failed row so a
+  current implementation re-schedules `fn` on a pending/failed/cancelled row so a
   duplicate caller may legitimately re-run it; once status is `done` all
   further calls short-circuit. See
   `TestRunAsyncMigration_ConcurrentSameNameSerialized` for the contract.

cmd/ingestor/async_migration.go

@@ -14,8 +14,8 @@
 //     1. Run via Store.RunAsyncMigration(...) below (preferred for backfills
 //        and any work that may touch >1K rows). The migration is recorded as
 //        `pending_async` immediately, returns to the caller (boot proceeds),
-//        and completes in a goroutine. Status flips to `done` (or `failed`
-//        with an error message) when fn returns.
+//        and completes in a goroutine. Status flips to `done`, `cancelled`,
+//        or `failed` when fn returns; interruptions retain the error message.
 //     2. Carry the preflight annotation comment immediately above the
 //        migration block, e.g.
 //             // PREFLIGHT: async=true reason="<one-line justification>"
@@ -32,6 +32,7 @@ package main
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"log"
 )
@@ -42,7 +43,7 @@ func ensureAsyncMigrationsTable(db *sql.DB) error {
 	_, err := db.Exec(`
 		CREATE TABLE IF NOT EXISTS _async_migrations (
 			name       TEXT PRIMARY KEY,
-			status     TEXT NOT NULL,             -- pending_async | done | failed
+			status     TEXT NOT NULL,             -- pending_async | done | cancelled | failed
 			started_at TEXT NOT NULL DEFAULT (datetime('now')),
 			ended_at   TEXT,
 			error      TEXT
@@ -57,10 +58,11 @@ func ensureAsyncMigrationsTable(db *sql.DB) error {
 //
 // Contract (pinned by async_migration_test.go):
 //   - status is `pending_async` IMMEDIATELY after this returns.
-//   - fn runs in a goroutine; on success status becomes `done`, on error or
-//     panic status becomes `failed` and the error is recorded.
+//   - fn runs in a goroutine; on success status becomes `done`. Errors matching
+//     context.Canceled become `cancelled`; other errors and panics become
+//     `failed`. Both interrupted states record the error and completion time.
 //   - Idempotent: if a row with the same name already exists in `done`
-//     state, fn is NOT re-run. If in `failed` or `pending_async` state,
+//     state, fn is NOT re-run. If in `failed`, `cancelled`, or `pending_async`,
 //     fn IS re-scheduled (a previous run may have crashed mid-flight).
 //   - The caller's WaitGroup tracks the goroutine so tests/shutdown can
 //     wait via Store.WaitForAsyncMigrations().
@@ -76,7 +78,7 @@ func (s *Store) RunAsyncMigration(ctx context.Context, name string, fn func(cont
 		if existing == "done" {
 			return nil // already complete, nothing to do
 		}
-		// pending_async or failed → reset and retry.
+		// pending_async, cancelled, or failed → reset and retry.
 		if _, err := s.db.Exec(`
 			UPDATE _async_migrations
 			SET status = 'pending_async', started_at = datetime('now'), ended_at = NULL, error = NULL
@@ -103,13 +105,17 @@ func (s *Store) RunAsyncMigration(ctx context.Context, name string, fn func(cont
 				log.Printf("[async-migration] %q panic recovered: %v", name, r)
 			}
 			if runErr != nil {
+				status, label := "failed", "FAILED"
+				if errors.Is(runErr, context.Canceled) {
+					status, label = "cancelled", "cancelled"
+				}
 				if _, err := s.db.Exec(`
 					UPDATE _async_migrations
-					SET status = 'failed', ended_at = datetime('now'), error = ?
-					WHERE name = ?`, runErr.Error(), name); err != nil {
-					log.Printf("[async-migration] failed to record failure for %q: %v", name, err)
+					SET status = ?, ended_at = datetime('now'), error = ?
+					WHERE name = ?`, status, runErr.Error(), name); err != nil {
+					log.Printf("[async-migration] failed to record %s for %q: %v", status, name, err)
 				}
-				log.Printf("[async-migration] %q FAILED: %v", name, runErr)
+				log.Printf("[async-migration] %q %s: %v", name, label, runErr)
 				return
 			}
 			if _, err := s.db.Exec(`
@@ -129,7 +135,7 @@ func (s *Store) RunAsyncMigration(ctx context.Context, name string, fn func(cont
 }
 
 // AsyncMigrationStatus returns the current status of an async migration
-// (one of "pending_async", "done", "failed") or sql.ErrNoRows if no such
+// (one of "pending_async", "done", "cancelled", "failed") or sql.ErrNoRows if no such
 // migration has been registered.
 func (s *Store) AsyncMigrationStatus(name string) (string, error) {
 	if err := ensureAsyncMigrationsTable(s.db); err != nil {

cmd/ingestor/async_migration_test.go

@@ -1,9 +1,12 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"fmt"
+	"log"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -149,43 +152,49 @@ func TestRunAsyncMigration_IdempotentSecondCallNoOps(t *testing.T) {
 	}
 }
 
-// TestRunAsyncMigration_RestartSafetyFailedIsRetried simulates a crashed
-// previous run: a row exists in `failed` state from a prior boot. The next
-// RunAsyncMigration call MUST re-schedule fn (reset to pending_async, then
-// run it), not leave the migration stuck in `failed` forever.
-func TestRunAsyncMigration_RestartSafetyFailedIsRetried(t *testing.T) {
-	s := newTestStore(t)
-	const name = "test_restart_failed_v1"
-
-	if err := ensureAsyncMigrationsTable(s.db); err != nil {
-		t.Fatalf("ensure table: %v", err)
-	}
-	if _, err := s.db.Exec(`INSERT INTO _async_migrations (name, status, error) VALUES (?, 'failed', 'simulated prior crash')`, name); err != nil {
-		t.Fatalf("seed failed row: %v", err)
-	}
-
-	var calls int32
-	if err := s.RunAsyncMigration(context.Background(), name,
-		func(ctx context.Context, db *sql.DB) error {
-			atomic.AddInt32(&calls, 1)
-			return nil
-		}); err != nil {
-		t.Fatalf("RunAsyncMigration on failed row: %v", err)
-	}
-	s.WaitForAsyncMigrations()
-	waitForStatus(t, s, name, "done", 2*time.Second)
-
-	if got := atomic.LoadInt32(&calls); got != 1 {
-		t.Fatalf("fn invoked %d times, want 1 (failed-state row must be retried)", got)
-	}
-
-	// And the error column must be cleared on success.
-	var errCol sql.NullString
-	if err := s.db.QueryRow(`SELECT error FROM _async_migrations WHERE name = ?`, name).Scan(&errCol); err != nil {
-		t.Fatalf("error col: %v", err)
-	}
-	if errCol.Valid && errCol.String != "" {
-		t.Fatalf("error column not cleared on retry success: %q", errCol.String)
+// Failed and cancelled runs are retried, with stale completion details cleared.
+func TestRunAsyncMigration_RestartSafetyInterruptedIsRetried(t *testing.T) {
+	for _, previous := range []string{"failed", "cancelled"} {
+		t.Run(previous, func(t *testing.T) {
+			s := newTestStore(t)
+			const name = "test_restart_interrupted_v1"
+			if _, err := s.db.Exec(`INSERT INTO _async_migrations (name, status, ended_at, error) VALUES (?, ?, '2026-01-01 00:00:00', 'prior interruption')`, name, previous); err != nil {
+				t.Fatalf("seed interrupted row: %v", err)
+			}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			t.Cleanup(func() { cancel(); s.WaitForAsyncMigrations() })
+			release := make(chan struct{})
+			var calls int32
+			if err := s.RunAsyncMigration(ctx, name, func(ctx context.Context, db *sql.DB) error {
+				atomic.AddInt32(&calls, 1)
+				select {
+				case <-release:
+					return nil
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			}); err != nil {
+				t.Fatalf("RunAsyncMigration on %s row: %v", previous, err)
+			}
+
+			var status string
+			var endedAt, errCol sql.NullString
+			if err := s.db.QueryRow(`SELECT status, ended_at, error FROM _async_migrations WHERE name = ?`, name).Scan(&status, &endedAt, &errCol); err != nil {
+				t.Fatal(err)
+			}
+			if status != "pending_async" || endedAt.Valid || errCol.Valid {
+				t.Fatalf("retry not reset: status=%q ended_at=%v error=%v", status, endedAt, errCol)
+			}
+			close(release)
+			s.WaitForAsyncMigrations()
+			if err := s.db.QueryRow(`SELECT status, ended_at, error FROM _async_migrations WHERE name = ?`, name).Scan(&status, &endedAt, &errCol); err != nil {
+				t.Fatal(err)
+			}
+			if status != "done" || !endedAt.Valid || endedAt.String == "" || errCol.Valid || atomic.LoadInt32(&calls) != 1 {
+				t.Fatalf("retry did not complete once: status=%q ended_at=%v error=%v calls=%d", status, endedAt, errCol, calls)
+			}
+		})
 	}
 }
 
@@ -220,34 +229,63 @@ func TestRunAsyncMigration_RestartSafetyPendingIsRetried(t *testing.T) {
 	}
 }
 
-// TestRunAsyncMigration_FnErrorRecorded covers the non-panic failure path:
-// fn returns an error → status MUST be "failed" with the error captured.
-func TestRunAsyncMigration_FnErrorRecorded(t *testing.T) {
-	s := newTestStore(t)
-	const name = "test_fn_error_v1"
-
-	if err := s.RunAsyncMigration(context.Background(), name,
-		func(ctx context.Context, db *sql.DB) error {
-			return fmt.Errorf("simulated migration error")
-		}); err != nil {
-		t.Fatalf("RunAsyncMigration: %v", err)
-	}
-	s.WaitForAsyncMigrations()
-
-	status, err := s.AsyncMigrationStatus(name)
-	if err != nil {
-		t.Fatalf("status: %v", err)
-	}
-	if status != "failed" {
-		t.Fatalf("status: got %q, want failed", status)
-	}
-
-	var errCol sql.NullString
-	if err := s.db.QueryRow(`SELECT error FROM _async_migrations WHERE name = ?`, name).Scan(&errCol); err != nil {
-		t.Fatalf("error col: %v", err)
-	}
-	if !errCol.Valid || errCol.String == "" {
-		t.Fatalf("error column empty after fn error")
+// Classify the returned error, not ctx.Err(): shutdown must not hide failures.
+func TestRunAsyncMigration_CompletionStatus(t *testing.T) {
+	for _, tt := range []struct {
+		name   string
+		result error
+		panic  bool
+		want   string
+	}{
+		{"cancelled", context.Canceled, false, "cancelled"},
+		{"wrapped cancellation", fmt.Errorf("backfill: %w", context.Canceled), false, "cancelled"},
+		{"ordinary error", fmt.Errorf("simulated migration error"), false, "failed"},
+		{"deadline exceeded", context.DeadlineExceeded, false, "failed"},
+		{"panic", context.Canceled, true, "failed"},
+		{"completed", nil, false, "done"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			s := newTestStore(t)
+			ctx, cancel := context.WithCancel(context.Background())
+			var logs bytes.Buffer
+			previousOutput := log.Writer()
+			log.SetOutput(&logs)
+			t.Cleanup(func() { cancel(); s.WaitForAsyncMigrations(); log.SetOutput(previousOutput) })
+			const name = "test_completion_v1"
+			if err := s.RunAsyncMigration(ctx, name, func(ctx context.Context, db *sql.DB) error {
+				<-ctx.Done()
+				if tt.panic {
+					panic(tt.result)
+				}
+				return tt.result
+			}); err != nil {
+				t.Fatal(err)
+			}
+			cancel()
+			s.WaitForAsyncMigrations()
+
+			var status string
+			var endedAt, errCol sql.NullString
+			if err := s.db.QueryRow(`SELECT status, ended_at, error FROM _async_migrations WHERE name = ?`, name).Scan(&status, &endedAt, &errCol); err != nil {
+				t.Fatal(err)
+			}
+			if status != tt.want || !endedAt.Valid || endedAt.String == "" {
+				t.Fatalf("completion: status=%q want=%q ended_at=%v", status, tt.want, endedAt)
+			}
+			wantErr := ""
+			if tt.result != nil {
+				wantErr = tt.result.Error()
+			}
+			if tt.panic {
+				wantErr = "panic: " + wantErr
+			}
+			if errCol.Valid != (tt.result != nil) || errCol.String != wantErr {
+				t.Fatalf("error=%v, want %q", errCol, wantErr)
+			}
+			if tt.want == "cancelled" && (strings.Contains(logs.String(), "FAILED") || !strings.Contains(logs.String(), "cancelled")) {
+				t.Fatalf("cancellation logged as failure: %s", logs.String())
+			}
+		})
 	}
 }
 

cmd/ingestor/stats_file_test.go

@@ -2,8 +2,11 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"os"
 	"path/filepath"
+	"runtime"
+	"syscall"
 	"testing"
 	"time"
 )
@@ -129,6 +132,9 @@ func TestWriteStatsAtomic_SymlinkAtDestIsReplaced(t *testing.T) {
 	// Pre-plant a symlink at the destination path.
 	path := filepath.Join(dir, "stats.json")
 	if err := os.Symlink(target, path); err != nil {
+		if runtime.GOOS == "windows" && errors.Is(err, syscall.Errno(1314)) {
+			t.Skip("Windows account cannot create symlinks; this security fixture must run on Linux CI")
+		}
 		t.Fatalf("symlink: %v", err)
 	}
 

cmd/server/area_filter_test.go

@@ -74,7 +74,7 @@ func TestGetNodePubkeysInArea_Polygon(t *testing.T) {
 }
 
 // newTestStoreWithDB builds a minimal PacketStore wired to the given DB and config.
-func newTestStoreWithDB(t *testing.T, db *DB, cfg *Config) *PacketStore {
+func newTestStoreWithDB(t testing.TB, db *DB, cfg *Config) *PacketStore {
 	t.Helper()
 	return &PacketStore{
 		db:                 db,