Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
0737f10
feat(notify): optional external node alerts feed and the node.externa…
efiten Oct 10, 2026
a5b60ca
test: make local validation independent of runner environment (#2160)
n30nex Oct 10, 2026
7f2c999
fix(ingestor): record cancelled background migrations separately (#2161)
n30nex Oct 10, 2026
be6aeff
perf(store): drop the duplicate subpath count map (#2163)
A13xB0 Oct 10, 2026
323b804
perf(frontend): throttle nav stats to one refresh per 15 s, none in h…
A13xB0 Oct 10, 2026
95672ac
perf(server): share /api/nodes responses for identical queries for 15…
A13xB0 Oct 10, 2026
8a3b923
perf(scope-stats): keep the per-region count on the first_seen index …
A13xB0 Oct 10, 2026
6969063
feat(scope-audit): transport view of every repeater seen forwarding (…
A13xB0 Oct 10, 2026
deb55dc
feat(scope-audit): transport view on the Scope Audit page (#2142, pag…
A13xB0 Oct 10, 2026
99efaed
fix(region-filter): setSelected notifies change listeners on a real c…
A13xB0 Oct 10, 2026
352a0cc
feat(region-filter): configurable region quick picks (#2170)
A13xB0 Oct 10, 2026
031ad9a
test(scope-audit): red — observer region filter for both views
A13xB0 Oct 10, 2026
037eb28
feat(scope-audit): observer region filter for both views
A13xB0 Oct 10, 2026
671f66c
test(scope-audit): red — the page sends the observer region to the API
A13xB0 Oct 10, 2026
3d17e86
feat(scope-audit): region filter on the page
A13xB0 Oct 10, 2026
6e9c31f
test(scope-audit): red — the region scan must keep the first_seen index
A13xB0 Oct 10, 2026
6cd10a8
perf(scope-audit): keep the region scan on the first_seen index
A13xB0 Oct 10, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/deploy.yml
Original file line number Diff line number Diff line change
Expand Up @@ -688,6 +688,7 @@ 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-table-sort.js
suite 1 test-touch-targets.js CHROMIUM_REQUIRE=1 BASE_URL=$B
suite 3 test-live-dedup.js BASE_URL=$B
Expand Down
11 changes: 7 additions & 4 deletions MIGRATIONS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
30 changes: 18 additions & 12 deletions cmd/ingestor/async_migration.go
Original file line number Diff line number Diff line change
Expand Up @@ -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>"
Expand All @@ -32,6 +32,7 @@ package main
import (
"context"
"database/sql"
"errors"
"fmt"
"log"
)
Expand All @@ -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
Expand All @@ -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().
Expand All @@ -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
Expand All @@ -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(`
Expand All @@ -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 {
Expand Down
168 changes: 103 additions & 65 deletions cmd/ingestor/async_migration_test.go
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
package main

import (
"bytes"
"context"
"database/sql"
"fmt"
"log"
"strings"
"sync"
"sync/atomic"
"testing"
Expand Down Expand Up @@ -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)
}
})
}
}

Expand Down Expand Up @@ -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())
}
})
}
}

Expand Down
6 changes: 6 additions & 0 deletions cmd/ingestor/stats_file_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,11 @@ package main

import (
"encoding/json"
"errors"
"os"
"path/filepath"
"runtime"
"syscall"
"testing"
"time"
)
Expand Down Expand Up @@ -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)
}

Expand Down
2 changes: 1 addition & 1 deletion cmd/server/area_filter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
1 change: 0 additions & 1 deletion cmd/server/channel_analytics_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ func newChannelTestStore(packets []*StoreTx) *PacketStore {
chanCache: make(map[string]*cachedResult),
distCache: make(map[string]*cachedResult),
subpathCache: make(map[string]*cachedResult),
spIndex: make(map[string]int),
spTxIndex: make(map[string][]*StoreTx),
advertPubkeys: make(map[string]int),
clockSkew: NewClockSkewEngine(),
Expand Down
Loading