mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-22 23:31:33 +00:00
planguard: plan and result regression checks, fixture tier in CI, read-only mode (#240)
* feat(planguard): read sqlc statement text and load a strict plan corpus * fix(planguard): fail on unextracted statements, keep fragments mode-neutral * feat(planguard): plan assertions, scoped sort exemptions and narrow retired-index verdicts * fix(planguard): an errored run always fails * fix(planguard): keep subplan conds off their parent scan; fail on unanalyzed budgets * feat(planguard): read-only plan runner, reports and the run command * fix(planguard): validate identifiers in emitted scripts; prove read-only runs * feat(planguard): end-state corpus, planted regressions and the generated cached-plan policy * fix(planguard): derive observer cases from data; mark legacy generic plans * fix(planguard): bound the dense-region route resolvers; test policy refusals * fix(planguard): re-measure the dense-region route cases; cover every policy refusal * feat(planguard): synthetic fixture built through the migrations, checked in CI * ci(planguard): build the fixture once; guard fixture data for identifiers * fix(planguard): hash every fixture generator input; harden the fixture guard * feat(planguard): snapshot diff over every column, named result classes and cursor walks * fix(planguard): bound the relaxed diff classes; derive the cursor tie from data * fix(planguard): refuse multi-statement text in the corpus runner * feat: read-only server mode and an HTTP page diff between two builds * fix(planguard): refuse escape strings and psql meta-commands in corpus text * fix: storediff follows each build's own cursor; prove read-only gates via the log * fix(planguard): match the PostgreSQL lexer on high-bit bytes; check every emitted line * feat(admin): daily read-only plan check on the maintenance pool, its metric and report * feat(planguard): daily prod check behind planguard.daily_check * fix(planguard): cancel fresh runs at shutdown; expose a never-ran check * ci(planguard): pin the fixture server and collation; refresh baselines * fix(planguard): calibrate route budgets on one dataset; skip entries whose index is absent
This commit is contained in:
MrAlders0n authored and GitHub committed
1 parent
dba66cada7
commit
94b4b40f9b
327 files changed
+35447
-79
No files matched your search
@@ -22,6 +22,17 @@ jobs:
|
||||
--health-interval 5s
|
||||
--health-timeout 5s
|
||||
--health-retries 10
|
||||
# The fixture tier's baselines were written on PostgreSQL 16.15 under C collation: keep both fixed here.
|
||||
planguard-postgres:
|
||||
image: postgres:16.15-alpine
|
||||
env:
|
||||
POSTGRES_PASSWORD: planguard-ci-only
|
||||
POSTGRES_DB: planguard
|
||||
POSTGRES_INITDB_ARGS: --locale=C --encoding=UTF8
|
||||
ports:
|
||||
- 5433:5432
|
||||
options: >-
|
||||
--health-cmd pg_isready --health-interval 5s --health-timeout 5s --health-retries 10
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
@@ -50,21 +61,35 @@ jobs:
|
||||
- name: Verify scope refresh concurrency
|
||||
run: go test -race ./internal/meshmapper ./internal/scopestore && go test -race ./db -run '^TestChannelIDCache'
|
||||
|
||||
- name: Verify backup startup against PostgreSQL 16
|
||||
- name: Verify backup startup and the read-only boot against PostgreSQL 16
|
||||
env:
|
||||
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
|
||||
run: go test ./cmd/beacon -run '^TestBackupConfigurationPostgres$' -count=1 -v
|
||||
run: go test ./cmd/beacon -run '^Test(BackupConfiguration|ReadOnlyBootWritesNothing)Postgres$' -count=1 -v
|
||||
|
||||
- name: Verify stats and endpoint queries against PostgreSQL 16
|
||||
env:
|
||||
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
|
||||
run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough)Postgres$' -count=1 -v
|
||||
run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites)Postgres$' -count=1 -v
|
||||
|
||||
- name: Verify route retention task against PostgreSQL 16
|
||||
env:
|
||||
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
|
||||
run: go test ./internal/background -run '^TestRouteRetentionTaskPostgres$' -count=1 -v
|
||||
|
||||
# Builds the fixture once; the planguard tests below reuse it (PLANGUARD_FIXTURE_DSN) after checking that its
|
||||
# marker records a build of the same files.
|
||||
- name: Planguard fixture tier (plain EXPLAIN on a migrated, skewed fixture)
|
||||
run: go run ./cmd/planguard run --tier fixture --dsn 'postgres://postgres:planguard-ci-only@127.0.0.1:5433/planguard?sslmode=disable' --out "$RUNNER_TEMP/planguard"
|
||||
|
||||
- name: Verify planguard against PostgreSQL 16
|
||||
env:
|
||||
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
|
||||
PLANGUARD_FIXTURE_DSN: postgres://postgres:planguard-ci-only@127.0.0.1:5433/planguard?sslmode=disable
|
||||
run: go test ./internal/planguard -timeout 20m -run '^Test(RunnerReadOnly|RunnerRecoversAfterError|UnresolvedParamsSkip|GenericModeShowsParams|CorpusStatementsPrepareClean|BuildFixtureThroughMigrations|BuildFixtureRefusesRealDatabase|BuildFixtureRebuildsStaleInputs|RunDiffSnapshot|RunDiffFindsReadmodelsSkip|StatementScannerMatches|PlanguardProdTier|PlanguardProdCancel|RequiresIndexSkips)Postgres$' -count=1 -v
|
||||
|
||||
- name: Plan policy up to date
|
||||
run: go run ./cmd/planguard gen-policy && git diff --exit-code db/planpolicy
|
||||
|
||||
- name: Verify backup command against PostgreSQL 16
|
||||
run: |
|
||||
CGO_ENABLED=0 go build -o "$RUNNER_TEMP/beacon-backup" ./cmd/beacon-backup
|
||||
|
||||
@@ -8,3 +8,4 @@ roadmap.md
|
||||
|
||||
binaries/
|
||||
cover.out
|
||||
planguard-out/
|
||||
+160
-63
@@ -18,6 +18,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/db"
|
||||
pgdata "github.com/MeshCore-Beacon/beacon-server/db/planguard"
|
||||
"github.com/MeshCore-Beacon/beacon-server/db/planpolicy"
|
||||
"github.com/MeshCore-Beacon/beacon-server/docs"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api"
|
||||
@@ -34,10 +35,13 @@ import (
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/keystore"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/logging"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/meshmapper"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/metrics"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/planguard"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/presence"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/joho/godotenv"
|
||||
)
|
||||
|
||||
@@ -139,7 +143,13 @@ func main() {
|
||||
|
||||
dsn := getEnv("POSTGRES_DSN")
|
||||
layout := resolved.APIPlanLayout
|
||||
pools, err := db.OpenPools(ctx, dsn, layout)
|
||||
mode := bootModeFor(resolved.ReadOnly)
|
||||
var tune []func(string, *pgxpool.Config)
|
||||
if resolved.ReadOnly {
|
||||
slog.Warn(readOnlyWarning, "component", "startup")
|
||||
tune = append(tune, readOnlyTune)
|
||||
}
|
||||
pools, err := db.OpenPools(ctx, dsn, layout, tune...)
|
||||
if err != nil {
|
||||
// Parse errors can embed the complete DSN, including its password.
|
||||
slog.Error("invalid PostgreSQL connection configuration; check POSTGRES_DSN", "component", "startup")
|
||||
@@ -149,9 +159,11 @@ func main() {
|
||||
slog.Info("database pools open", "component", "startup", "layout", layout, "connections", pools.Total())
|
||||
backupOpts := configureBackup(ctx, cfg, pools.Maint, dsn, configPath)
|
||||
|
||||
if err := db.RunMigrations(ctx, pools.Maint); err != nil {
|
||||
slog.Error("migrations failed", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
if mode.Migrate {
|
||||
if err := db.RunMigrations(ctx, pools.Maint); err != nil {
|
||||
slog.Error("migrations failed", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// Three stores: ingest writes, background and operator work, and API reads. The API store
|
||||
@@ -172,7 +184,9 @@ func main() {
|
||||
// ── Presence write coalescing ────────────────────────────────────────────
|
||||
// Ingest writes go through the coalescer; reads use the API store.
|
||||
coalescer := presence.New(ingestStore, resolved.PresenceFlushInterval, resolved.PresencePacketTTL)
|
||||
go coalescer.Run(ctx)
|
||||
if mode.Coalescer {
|
||||
go coalescer.Run(ctx)
|
||||
}
|
||||
|
||||
// ── Redis cache layer ────────────────────────────────────────────────────
|
||||
var reader api.Reader = apiStore
|
||||
@@ -201,9 +215,11 @@ func main() {
|
||||
}
|
||||
|
||||
// ── Seed config data ─────────────────────────────────────────────────────
|
||||
if err := config.Seed(ctx, cfg, maintStore); err != nil {
|
||||
slog.Error("failed to seed config", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
if mode.Seed {
|
||||
if err := config.Seed(ctx, cfg, maintStore); err != nil {
|
||||
slog.Error("failed to seed config", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// ── Build transport scope keystore ───────────────────────────────────────
|
||||
@@ -217,7 +233,7 @@ func main() {
|
||||
scopes.SetManualMembers(cfg.ManualScopeMembers())
|
||||
slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup")
|
||||
directory := meshmapper.NewDirectory(maintStore, cfg.MeshMapper.APIKey)
|
||||
if mm := cfg.MeshMapper; mm.Scopes.Enabled || mm.Zones.Enabled || mm.Channels.Enabled || mm.Repeaters.Enabled {
|
||||
if mm := cfg.MeshMapper; mode.MeshMapper && (mm.Scopes.Enabled || mm.Zones.Enabled || mm.Channels.Enabled || mm.Repeaters.Enabled) {
|
||||
if cfg.MeshMapper.APIKey == "" {
|
||||
slog.Warn("MeshMapper integration unconfigured: set MESHMAPPER_API_KEY or meshmapper.api_key; retaining cached data", "component", "meshmapper")
|
||||
}
|
||||
@@ -230,7 +246,7 @@ func main() {
|
||||
}
|
||||
}
|
||||
var scopeImporter *meshmapper.Importer
|
||||
if cfg.MeshMapper.Scopes.Enabled {
|
||||
if mode.MeshMapper && cfg.MeshMapper.Scopes.Enabled {
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
scopeImporter, err = meshmapper.New(restoreCtx, cfg.MeshMapper.Scopes, maintStore, directory, scopes, scopeEntries)
|
||||
cancelRestore()
|
||||
@@ -264,12 +280,15 @@ func main() {
|
||||
localBorders.Store(local)
|
||||
})
|
||||
}
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
err = zones.Restore(restoreCtx)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper boundaries", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
// Restore prunes boundaries a disabled import left behind, so a read-only server skips it.
|
||||
if mode.MeshMapper {
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
err = zones.Restore(restoreCtx)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper boundaries", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
if localBorders != nil && !localBorders.Ready() {
|
||||
slog.Warn("nodes.mark_foreign has no boundaries yet; waiting for MeshMapper", "component", "startup")
|
||||
@@ -315,47 +334,56 @@ func main() {
|
||||
|
||||
keys := keystore.NewMapKeyStore(entries)
|
||||
|
||||
// Restored before the boot backfill so saved MeshMapper keys decrypt history too.
|
||||
restoreCtx, cancelRestore = context.WithTimeout(ctx, 10*time.Second)
|
||||
channelImporter, err := meshmapper.NewChannels(restoreCtx, cfg.MeshMapper.Channels, maintStore, directory, keys)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper channels", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
// Restored before the boot backfill so saved MeshMapper keys decrypt history too. Both
|
||||
// constructors may clear rows when their import is disabled, so a read-only server skips them.
|
||||
var channelImporter *meshmapper.Channels
|
||||
var repeaterImporter *meshmapper.Repeaters
|
||||
if mode.MeshMapper {
|
||||
restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second)
|
||||
channelImporter, err = meshmapper.NewChannels(restoreCtx, cfg.MeshMapper.Channels, maintStore, directory, keys)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper channels", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var onRepeatersChange func(context.Context)
|
||||
if cr, ok := reader.(*cache.CachedReader); ok {
|
||||
onRepeatersChange = cr.InvalidateNodeDetails
|
||||
}
|
||||
restoreCtx, cancelRestore = context.WithTimeout(ctx, 10*time.Second)
|
||||
repeaterImporter, err = meshmapper.NewRepeaters(restoreCtx, cfg.MeshMapper.Repeaters, maintStore, directory, onRepeatersChange)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper repeaters", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
var onRepeatersChange func(context.Context)
|
||||
if cr, ok := reader.(*cache.CachedReader); ok {
|
||||
onRepeatersChange = cr.InvalidateNodeDetails
|
||||
if channelImporter != nil && mode.Backfill {
|
||||
channelImporter.OnNewKeys(func(hashes [][]byte) {
|
||||
// Off the refresh tick: a backfill can outlast it, and history isn't urgent.
|
||||
go func() {
|
||||
backfillCtx, cancel := context.WithTimeout(ctx, 10*time.Minute)
|
||||
defer cancel()
|
||||
if n, err := ingest.BackfillChannelHashes(backfillCtx, maintStore, keys, hashes); err != nil {
|
||||
slog.Error("MeshMapper channel backfill failed", "component", "meshmapper.channels", "error", err)
|
||||
} else if n > 0 {
|
||||
slog.Info(fmt.Sprintf("backfilled %d channel message(s) for imported keys", n), "component", "meshmapper.channels")
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
restoreCtx, cancelRestore = context.WithTimeout(ctx, 10*time.Second)
|
||||
repeaterImporter, err := meshmapper.NewRepeaters(restoreCtx, cfg.MeshMapper.Repeaters, maintStore, directory, onRepeatersChange)
|
||||
cancelRestore()
|
||||
if err != nil {
|
||||
slog.Error("failed to restore MeshMapper repeaters", "component", "startup", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
channelImporter.OnNewKeys(func(hashes [][]byte) {
|
||||
// Off the refresh tick: a backfill can outlast it, and history isn't urgent.
|
||||
go func() {
|
||||
backfillCtx, cancel := context.WithTimeout(ctx, 10*time.Minute)
|
||||
defer cancel()
|
||||
if n, err := ingest.BackfillChannelHashes(backfillCtx, maintStore, keys, hashes); err != nil {
|
||||
slog.Error("MeshMapper channel backfill failed", "component", "meshmapper.channels", "error", err)
|
||||
} else if n > 0 {
|
||||
slog.Info(fmt.Sprintf("backfilled %d channel message(s) for imported keys", n), "component", "meshmapper.channels")
|
||||
}
|
||||
}()
|
||||
})
|
||||
|
||||
// ── Backfill channel messages ────────────────────────────────────────────
|
||||
// Packets whose channel key wasn't yet configured at ingest time were stored as
|
||||
// hash-only channels and never decrypted. Retry them now against the keystore we just
|
||||
// built, so adding a channel key to the config surfaces its history on the next boot
|
||||
// instead of leaving it stranded in the DB indefinitely.
|
||||
if n, err := ingest.BackfillChannelMessages(ctx, maintStore, keys); err != nil {
|
||||
slog.Error("config: channel message backfill failed", "component", "startup", "error", err)
|
||||
} else if n > 0 {
|
||||
slog.Info(fmt.Sprintf("config: backfilled %d previously-undecrypted channel message(s)", n), "component", "startup")
|
||||
if mode.Backfill {
|
||||
if n, err := ingest.BackfillChannelMessages(ctx, maintStore, keys); err != nil {
|
||||
slog.Error("config: channel message backfill failed", "component", "startup", "error", err)
|
||||
} else if n > 0 {
|
||||
slog.Info(fmt.Sprintf("config: backfilled %d previously-undecrypted channel message(s)", n), "component", "startup")
|
||||
}
|
||||
}
|
||||
|
||||
// ── Build geographic ingest filter ───────────────────────────────────────────────────────────
|
||||
@@ -407,15 +435,19 @@ func main() {
|
||||
}
|
||||
|
||||
var ingestWorkers sync.WaitGroup
|
||||
ingestWorkers.Go(func() { broker1.Start(ctx) })
|
||||
ingestWorkers.Go(func() { broker2.Start(ctx) })
|
||||
if mode.Ingest {
|
||||
ingestWorkers.Go(func() { broker1.Start(ctx) })
|
||||
ingestWorkers.Go(func() { broker2.Start(ctx) })
|
||||
}
|
||||
|
||||
rollup := background.NewRollup(maintStore, resolved.PacketRetention, resolved.RollupRetention)
|
||||
go func() {
|
||||
if err := rollup.CatchUp(ctx); err != nil {
|
||||
slog.Error("analytics rollup catch-up failed", "component", "background", "error", err)
|
||||
}
|
||||
}()
|
||||
if mode.Jobs {
|
||||
go func() {
|
||||
if err := rollup.CatchUp(ctx); err != nil {
|
||||
slog.Error("analytics rollup catch-up failed", "component", "background", "error", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
tasks := []background.Task{
|
||||
rollup.Task(),
|
||||
background.ViewRefreshTask(maintStore, resolved.ViewRefreshInterval),
|
||||
@@ -441,22 +473,28 @@ func main() {
|
||||
if scopeImporter != nil {
|
||||
tasks = append(tasks, background.Task{Name: "meshmapper.scopes", Interval: meshmapper.PollInterval, Run: scopeImporter.Refresh})
|
||||
}
|
||||
if cfg.MeshMapper.Zones.Enabled {
|
||||
if mode.MeshMapper && cfg.MeshMapper.Zones.Enabled {
|
||||
tasks = append(tasks, background.Task{Name: "meshmapper.zones", Interval: meshmapper.PollInterval, Run: zones.Refresh})
|
||||
}
|
||||
if cfg.MeshMapper.Channels.Enabled {
|
||||
if channelImporter != nil && cfg.MeshMapper.Channels.Enabled {
|
||||
tasks = append(tasks, background.Task{Name: "meshmapper.channels", Interval: meshmapper.PollInterval, Run: channelImporter.Refresh})
|
||||
}
|
||||
if cfg.MeshMapper.Repeaters.Enabled {
|
||||
if repeaterImporter != nil && cfg.MeshMapper.Repeaters.Enabled {
|
||||
tasks = append(tasks, background.Task{Name: "meshmapper.repeaters", Interval: meshmapper.PollInterval, Run: repeaterImporter.Refresh})
|
||||
}
|
||||
pgSource, pgTask := planguardWiring(ctx, resolved.PlanguardDailyCheck, mode.Jobs, pools.Maint)
|
||||
if pgTask != nil {
|
||||
tasks = append(tasks, *pgTask)
|
||||
}
|
||||
profiles := configureProfiling(ctx, pools.Maint)
|
||||
defer profiles.Stop()
|
||||
for i := range tasks {
|
||||
tasks[i].Run = profiles.WrapTask(tasks[i].Name, tasks[i].Run)
|
||||
}
|
||||
scheduler := background.New(tasks)
|
||||
go scheduler.Start(ctx)
|
||||
if mode.Jobs {
|
||||
scheduler := background.New(tasks)
|
||||
go scheduler.Start(ctx)
|
||||
}
|
||||
|
||||
// ── HTTP server ──────────────────────────────────────────────────────────
|
||||
// Wrap after wiring cache invalidators and cleanup callbacks to the actual
|
||||
@@ -474,7 +512,7 @@ func main() {
|
||||
AdminRoutes: map[string]http.Handler{
|
||||
"/accounts": handlers.AccountsRouter(maintStore),
|
||||
"/backup": handlers.BackupRouter(backupOpts, ctx),
|
||||
"/db": handlers.DBAdminRouter(maintStore),
|
||||
"/db": handlers.DBAdminRouter(maintStore, pgSource),
|
||||
},
|
||||
Guards: guard,
|
||||
APIPlanLayout: string(layout),
|
||||
@@ -520,7 +558,9 @@ func main() {
|
||||
}
|
||||
}
|
||||
ingestWorkers.Wait()
|
||||
coalescer.Flush(shutdownCtx)
|
||||
if mode.Coalescer {
|
||||
coalescer.Flush(shutdownCtx)
|
||||
}
|
||||
}
|
||||
|
||||
// proxyLimitWarning explains why per-IP limits go site-wide behind an unlisted reverse proxy.
|
||||
@@ -545,3 +585,60 @@ func getEnv(key string) string {
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// planguardWiring builds the daily read-only plan check on the maintenance pool, behind planguard.daily_check.
|
||||
// ctx is the server's root context: its end cancels any run in flight.
|
||||
// Off, /planguard answers that the check is disabled. On, the check runs with the other jobs (config validation
|
||||
// refuses it on a read-only server); a corpus that does not load disables it but never stops the server, and the
|
||||
// source stays a nil interface, so /planguard is 404.
|
||||
func planguardWiring(ctx context.Context, enabled, jobs bool, maint *pgxpool.Pool) (handlers.PlanguardSource, *background.Task) {
|
||||
if !enabled {
|
||||
return disabledPlanguard{}, nil
|
||||
}
|
||||
// On, the last-run timestamp is exported from boot (0 until a run succeeds), so a check that never succeeds,
|
||||
// even one whose corpus does not load, still trips the staleness alert.
|
||||
metrics.EnablePlanguard()
|
||||
// Defensive: configuration refuses planguard.daily_check together with server.read_only, the only mode
|
||||
// without jobs, so this branch is not reached today. It keeps a jobless server from ever running the check.
|
||||
if !jobs {
|
||||
return nil, nil
|
||||
}
|
||||
checker, err := planguard.NewProdChecker(maint, db.SQLCSources, pgdata.Files)
|
||||
if err != nil {
|
||||
slog.Error("plan check disabled", "component", "startup", "error", err)
|
||||
return nil, nil
|
||||
}
|
||||
pg := background.NewPlanguard(ctx, checker.Run)
|
||||
task := pg.Task(24 * time.Hour)
|
||||
return planguardSource{pg}, &task
|
||||
}
|
||||
|
||||
// planguardSource serves the plan check's reports in their API shape.
|
||||
type planguardSource struct{ pg *background.Planguard }
|
||||
|
||||
func (s planguardSource) Last() (api.PlanguardReport, error) {
|
||||
rep, ok := s.pg.Last()
|
||||
if !ok {
|
||||
return api.PlanguardReport{}, api.ErrPlanguardNotRun
|
||||
}
|
||||
return rep.API(), nil
|
||||
}
|
||||
|
||||
func (s planguardSource) RunNow(ctx context.Context) (api.PlanguardReport, error) {
|
||||
rep, err := s.pg.RunNow(ctx)
|
||||
if err != nil {
|
||||
return api.PlanguardReport{}, err
|
||||
}
|
||||
return rep.API(), nil
|
||||
}
|
||||
|
||||
// disabledPlanguard answers /planguard when planguard.daily_check is off.
|
||||
type disabledPlanguard struct{}
|
||||
|
||||
func (disabledPlanguard) Last() (api.PlanguardReport, error) {
|
||||
return api.PlanguardReport{}, api.ErrPlanguardDisabled
|
||||
}
|
||||
|
||||
func (disabledPlanguard) RunNow(context.Context) (api.PlanguardReport, error) {
|
||||
return api.PlanguardReport{}, api.ErrPlanguardDisabled
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api/handlers"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/metrics"
|
||||
"github.com/prometheus/client_golang/prometheus/testutil"
|
||||
)
|
||||
|
||||
// lastRunSeries counts the beacon_planguard_last_run_timestamp_seconds series the registry exports.
|
||||
func lastRunSeries(t *testing.T) int {
|
||||
t.Helper()
|
||||
n, err := testutil.GatherAndCount(metrics.Registry, "beacon_planguard_last_run_timestamp_seconds")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func TestPlanguardWiring(t *testing.T) {
|
||||
get := func(t *testing.T, src handlers.PlanguardSource, target string) (int, string) {
|
||||
t.Helper()
|
||||
w := httptest.NewRecorder()
|
||||
handlers.DBAdminRouter(nil, src).ServeHTTP(w, httptest.NewRequest("GET", target, nil))
|
||||
var body map[string]handlers.APIError
|
||||
_ = json.Unmarshal(w.Body.Bytes(), &body)
|
||||
return w.Code, body["error"].Message
|
||||
}
|
||||
|
||||
// The default config leaves the check off: no task, and /planguard says so, fresh or not.
|
||||
if config.Resolve(&config.Config{}).PlanguardDailyCheck {
|
||||
t.Fatal("planguard.daily_check is on by default")
|
||||
}
|
||||
src, task := planguardWiring(context.Background(), false, true, nil)
|
||||
if task != nil {
|
||||
t.Errorf("disabled check scheduled a task %q", task.Name)
|
||||
}
|
||||
for _, target := range []string{"/planguard", "/planguard?fresh=true"} {
|
||||
if code, msg := get(t, src, target); code != 404 || msg != "planguard daily check is disabled" {
|
||||
t.Errorf("disabled %s: %d %q", target, code, msg)
|
||||
}
|
||||
}
|
||||
|
||||
if n := lastRunSeries(t); n != 0 {
|
||||
t.Errorf("disabled check exports %d last-run series, want 0", n)
|
||||
}
|
||||
|
||||
// On: a daily task on the hourly tick, and 404 "has not run yet" until its first run.
|
||||
src, task = planguardWiring(context.Background(), true, true, nil)
|
||||
if task == nil || task.Name != "planguard" || task.Interval.Hours() != 1 {
|
||||
t.Fatalf("enabled check: task %+v", task)
|
||||
}
|
||||
if code, msg := get(t, src, "/planguard"); code != 404 || msg != "planguard has not run yet" {
|
||||
t.Errorf("enabled before the first run: %d %q", code, msg)
|
||||
}
|
||||
if n := lastRunSeries(t); n != 1 {
|
||||
t.Errorf("enabled check exports %d last-run series before its first run, want 1 (at 0)", n)
|
||||
}
|
||||
|
||||
// On without jobs: no task and no route.
|
||||
src, task = planguardWiring(context.Background(), true, false, nil)
|
||||
if task != nil || src != nil {
|
||||
t.Errorf("enabled without jobs: task %v source %v", task, src)
|
||||
}
|
||||
if code, _ := get(t, src, "/planguard"); code != 404 {
|
||||
t.Errorf("enabled without jobs: %d", code)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package main
|
||||
|
||||
import "github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
// readOnlyWarning is logged once at boot when server.read_only is set.
|
||||
const readOnlyWarning = "server.read_only: no migrations, ingest, background jobs or MeshMapper imports; every database session is read-only"
|
||||
|
||||
// bootMode says which boot steps that write may run. Each field gates one group of call sites in main.
|
||||
type bootMode struct {
|
||||
// Migrate runs the schema migrations.
|
||||
Migrate bool
|
||||
// Seed writes the configured IATAs, regions and scopes.
|
||||
Seed bool
|
||||
// Backfill decrypts stored channel messages at boot and for newly imported keys.
|
||||
Backfill bool
|
||||
// Ingest starts the MQTT ingest workers.
|
||||
Ingest bool
|
||||
// Coalescer runs and flushes the presence write coalescer.
|
||||
Coalescer bool
|
||||
// Jobs runs the rollup catch-up and the background scheduler.
|
||||
Jobs bool
|
||||
// MeshMapper restores and runs the MeshMapper importers; their restores can prune or clear rows.
|
||||
MeshMapper bool
|
||||
}
|
||||
|
||||
// bootModeFor returns every step for a normal server and none for a read-only one.
|
||||
func bootModeFor(readOnly bool) bootMode {
|
||||
on := !readOnly
|
||||
return bootMode{Migrate: on, Seed: on, Backfill: on, Ingest: on, Coalescer: on, Jobs: on, MeshMapper: on}
|
||||
}
|
||||
|
||||
// readOnlyTune makes every session of a pool read-only, so a write fails with SQLSTATE 25006
|
||||
// instead of changing the database. It is passed to db.OpenPools when server.read_only is set.
|
||||
func readOnlyTune(_ string, cfg *pgxpool.Config) {
|
||||
cfg.ConnConfig.RuntimeParams["default_transaction_read_only"] = "on"
|
||||
}
|
||||
@@ -0,0 +1,411 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
//go:build unix
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"io"
|
||||
"maps"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/db"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
const readOnlyBootAPIKey = "read-only-boot-test-key"
|
||||
|
||||
// readOnlyBootConfig makes a normal boot write: Seed upserts the IATA, region and hashtag
|
||||
// channel, the disabled MeshMapper imports clear their rows, and the jobs run at once.
|
||||
const readOnlyBootConfig = `server:
|
||||
read_only: %t
|
||||
ratelimit:
|
||||
enabled: false
|
||||
auth:
|
||||
api_key: ` + readOnlyBootAPIKey + `
|
||||
iatas:
|
||||
SEA:
|
||||
name: Seattle
|
||||
lat: 47.45
|
||||
lng: -122.31
|
||||
regions:
|
||||
- name: Pacific Northwest
|
||||
slug: pnw
|
||||
iatas: [SEA]
|
||||
channel_keys:
|
||||
hashtags:
|
||||
- readonlytest
|
||||
`
|
||||
|
||||
// readOnlyRequests is the request mix a read-only server serves. GETs must not fail with a
|
||||
// server error; writes must fail and never succeed.
|
||||
var readOnlyRequests = []struct {
|
||||
method, path, body string
|
||||
write bool
|
||||
}{
|
||||
{"GET", "/api/v1/info", "", false},
|
||||
{"GET", "/api/v1/iatas", "", false},
|
||||
{"GET", "/api/v1/iatas/SEA", "", false},
|
||||
{"GET", "/api/v1/regions", "", false},
|
||||
{"GET", "/api/v1/nodes?limit=5", "", false},
|
||||
{"GET", "/api/v1/observers?limit=5", "", false},
|
||||
{"GET", "/api/v1/observers/directory", "", false},
|
||||
{"GET", "/api/v1/packets?limit=5", "", false},
|
||||
{"GET", "/api/v1/routes?limit=5", "", false},
|
||||
{"GET", "/api/v1/routes?iata=SEA&limit=5", "", false},
|
||||
{"GET", "/api/v1/channels", "", false},
|
||||
{"GET", "/api/v1/messages?limit=5", "", false},
|
||||
{"GET", "/api/v1/scopes", "", false},
|
||||
{"GET", "/api/v1/brokers", "", false},
|
||||
{"GET", "/api/v1/traces", "", false},
|
||||
{"GET", "/api/v1/stats/overview", "", false},
|
||||
{"GET", "/api/v1/stats/node-types", "", false},
|
||||
{"GET", "/api/v1/stats/top-nodes", "", false},
|
||||
{"GET", "/api/v1/admin/accounts", "", false},
|
||||
{"POST", "/api/v1/admin/accounts", `{"name":"read-only boot"}`, true},
|
||||
}
|
||||
|
||||
// tableStat is a table's live row count and its insert, update and delete counters.
|
||||
type tableStat struct{ rows, ins, upd, del int64 }
|
||||
|
||||
// A read-only server boots against a migrated database, serves a request mix and shuts down
|
||||
// without changing any table: no row count and no insert, update or delete counter moves.
|
||||
// The same boot without read_only does move them, so the check can see a write.
|
||||
func TestReadOnlyBootWritesNothingPostgres(t *testing.T) {
|
||||
if os.Getenv("BEACON_TEST_BOOT_MAIN") == "1" {
|
||||
// The parent holds stdin open; when it dies, even by SIGKILL, stdin closes and the server goes with it.
|
||||
go func() {
|
||||
_, _ = io.Copy(io.Discard, os.Stdin)
|
||||
os.Exit(3)
|
||||
}()
|
||||
main()
|
||||
return
|
||||
}
|
||||
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("set BEACON_TEST_POSTGRES_DSN")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
admin, err := pgx.Connect(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
name := "beacon_read_only_" + strings.ToLower(rand.Text())
|
||||
ident := pgx.Identifier{name}.Sanitize()
|
||||
_, err = admin.Exec(ctx, "CREATE DATABASE "+ident)
|
||||
admin.Close(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
if admin, err := pgx.Connect(ctx, dsn); err == nil {
|
||||
_, _ = admin.Exec(ctx, "DROP DATABASE "+ident+" WITH (FORCE)")
|
||||
admin.Close(ctx)
|
||||
}
|
||||
})
|
||||
cfg, err := pgxpool.ParseConfig(dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg.ConnConfig.Database = name
|
||||
pool, err := pgxpool.NewWithConfig(ctx, cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.RunMigrations(ctx, pool); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, q := range []string{
|
||||
"INSERT INTO iata_codes(iata) VALUES ('YVR')",
|
||||
"INSERT INTO observers(id, public_key) VALUES ('00000000-0000-0000-0000-000000000001', '\\x01')",
|
||||
"INSERT INTO nodes(id, public_key, name, node_type) VALUES ('00000000-0000-0000-0000-0000000000a1', '\\xaa11', 'Alice', 2)",
|
||||
} {
|
||||
if _, err := pool.Exec(ctx, q); err != nil {
|
||||
t.Fatalf("%s: %v", q, err)
|
||||
}
|
||||
}
|
||||
pool.Close()
|
||||
statsConn := func() *pgx.Conn {
|
||||
c, err := pgx.ConnectConfig(ctx, cfg.ConnConfig.Copy())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// The control boot writes; it also leaves the database as a server would after a normal start.
|
||||
before := snapshotTables(t, ctx, statsConn(), name)
|
||||
bootAndServe(t, ctx, poolDSN(t, dsn, name), false)
|
||||
after := snapshotTables(t, ctx, statsConn(), name)
|
||||
if maps.Equal(before, after) {
|
||||
t.Fatal("a normal boot changed no table, so the check cannot see a write")
|
||||
}
|
||||
|
||||
before = after
|
||||
log := bootAndServe(t, ctx, poolDSN(t, dsn, name), true)
|
||||
after = snapshotTables(t, ctx, statsConn(), name)
|
||||
for _, table := range slices.Sorted(maps.Keys(after)) {
|
||||
if before[table] != after[table] {
|
||||
t.Errorf("%s changed under read_only: before %+v, after %+v", table, before[table], after[table])
|
||||
}
|
||||
}
|
||||
if len(before) != len(after) {
|
||||
t.Errorf("table set changed under read_only: %d tables before, %d after", len(before), len(after))
|
||||
}
|
||||
if !strings.Contains(log, readOnlyWarning) {
|
||||
t.Errorf("boot log lacks the read-only warning:\n%s", log)
|
||||
}
|
||||
// A gated step that still ran would have hit a read-only session and logged it. The only refused write is
|
||||
// the request mix's admin POST.
|
||||
refused := 0
|
||||
for _, line := range strings.Split(log, "\n") {
|
||||
if !strings.Contains(line, "25006") && !strings.Contains(line, "read-only transaction") {
|
||||
continue
|
||||
}
|
||||
if strings.Contains(line, `msg="account request failed"`) {
|
||||
refused++
|
||||
continue
|
||||
}
|
||||
t.Errorf("read-only server refused a write outside the request mix: %s", line)
|
||||
}
|
||||
if refused != 1 {
|
||||
t.Errorf("the admin POST logged %d refused writes, want 1:\n%s", refused, log)
|
||||
}
|
||||
}
|
||||
|
||||
// poolDSN points dsn at the named database.
|
||||
func poolDSN(t *testing.T, dsn, name string) string {
|
||||
t.Helper()
|
||||
cfg, err := pgx.ParseConfig(dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return fmt.Sprintf("postgres://%s@%s/%s?sslmode=disable", userInfo(cfg), net.JoinHostPort(cfg.Host, fmt.Sprint(cfg.Port)), name)
|
||||
}
|
||||
|
||||
func userInfo(cfg *pgx.ConnConfig) string {
|
||||
if cfg.Password == "" {
|
||||
return cfg.User
|
||||
}
|
||||
return cfg.User + ":" + cfg.Password
|
||||
}
|
||||
|
||||
// bootAndServe runs the server in a child process, sends the request mix once it is ready,
|
||||
// stops it with SIGTERM and returns its log.
|
||||
func bootAndServe(t *testing.T, ctx context.Context, dsn string, readOnly bool) string {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
if err := os.WriteFile(path, []byte(fmt.Sprintf(readOnlyBootConfig, readOnly)), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
addr := l.Addr().String()
|
||||
l.Close()
|
||||
|
||||
cmd := exec.Command(os.Args[0], "-test.run=^TestReadOnlyBootWritesNothingPostgres$")
|
||||
cmd.Dir = dir
|
||||
cmd.Env = append(os.Environ(), "BEACON_TEST_BOOT_MAIN=1", "CONFIG_PATH="+path, "LISTEN_ADDR="+addr,
|
||||
"POSTGRES_DSN="+dsn, "BEACON_API_KEY="+readOnlyBootAPIKey, "MESHMAPPER_API_KEY=", "REDIS_ADDR=",
|
||||
"LOG_FORMAT=text", "LOG_LEVEL=info")
|
||||
var out lockedBuffer
|
||||
cmd.Stdout, cmd.Stderr = &out, &out
|
||||
// Its own process group, so cleanup reaches anything it started.
|
||||
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
|
||||
stdin, err := cmd.StdinPipe()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := cmd.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- cmd.Wait() }()
|
||||
stopped := false
|
||||
defer func() {
|
||||
if !stopped {
|
||||
_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
|
||||
<-done
|
||||
}
|
||||
stdin.Close()
|
||||
}()
|
||||
|
||||
base := "http://" + addr
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
ready := time.Now().Add(60 * time.Second)
|
||||
for {
|
||||
resp, err := client.Get(base + "/api/v1/info")
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
break
|
||||
}
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
stopped = true
|
||||
t.Fatalf("server exited before it was ready (read_only=%t): %v\n%s", readOnly, err, out.String())
|
||||
default:
|
||||
}
|
||||
if time.Now().After(ready) {
|
||||
t.Fatalf("server not ready (read_only=%t):\n%s", readOnly, out.String())
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
|
||||
if readOnly {
|
||||
for _, r := range readOnlyRequests {
|
||||
req, err := http.NewRequestWithContext(ctx, r.method, base+r.path, strings.NewReader(r.body))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+readOnlyBootAPIKey)
|
||||
if r.body != "" {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("%s %s: %v", r.method, r.path, err)
|
||||
}
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
switch {
|
||||
case r.write && resp.StatusCode < 300:
|
||||
t.Errorf("%s %s succeeded under read_only: %d %s", r.method, r.path, resp.StatusCode, body)
|
||||
case !r.write && resp.StatusCode >= 500:
|
||||
t.Errorf("%s %s: %d %s", r.method, r.path, resp.StatusCode, body)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Give the boot-time jobs a moment to write before the control stops.
|
||||
time.Sleep(2 * time.Second)
|
||||
}
|
||||
|
||||
if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
stopped = true
|
||||
if err != nil {
|
||||
t.Fatalf("server exit (read_only=%t): %v\n%s", readOnly, err, out.String())
|
||||
}
|
||||
case <-time.After(60 * time.Second):
|
||||
t.Fatalf("server did not stop (read_only=%t):\n%s", readOnly, out.String())
|
||||
}
|
||||
return out.String()
|
||||
}
|
||||
|
||||
// snapshotTables reads every user table's row count and tuple counters once the server's
|
||||
// sessions have ended; a backend flushes its statistics as it exits.
|
||||
func snapshotTables(t *testing.T, ctx context.Context, conn *pgx.Conn, database string) map[string]tableStat {
|
||||
t.Helper()
|
||||
defer conn.Close(context.Background())
|
||||
waitSessionsGone(t, ctx, conn, database)
|
||||
// Statistics arrive as backends flush them; read until two snapshots in a row agree.
|
||||
prev := readTables(t, ctx, conn)
|
||||
for deadline := time.Now().Add(30 * time.Second); ; {
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
cur := readTables(t, ctx, conn)
|
||||
if maps.Equal(prev, cur) {
|
||||
return cur
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("table statistics did not settle")
|
||||
}
|
||||
prev = cur
|
||||
}
|
||||
}
|
||||
|
||||
func waitSessionsGone(t *testing.T, ctx context.Context, conn *pgx.Conn, database string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(30 * time.Second)
|
||||
for {
|
||||
var n int
|
||||
if err := conn.QueryRow(ctx, "SELECT count(*) FROM pg_stat_activity WHERE datname = $1 AND application_name LIKE 'beacon-%'", database).Scan(&n); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n == 0 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("%d server sessions still open", n)
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// readTables reads every user table's row count and tuple counters.
|
||||
func readTables(t *testing.T, ctx context.Context, conn *pgx.Conn) map[string]tableStat {
|
||||
t.Helper()
|
||||
if _, err := conn.Exec(ctx, "SELECT pg_stat_clear_snapshot()"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows, err := conn.Query(ctx, "SELECT schemaname, relname, n_tup_ins, n_tup_upd, n_tup_del FROM pg_stat_user_tables")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stats := map[string]tableStat{}
|
||||
for rows.Next() {
|
||||
var schema, rel string
|
||||
var s tableStat
|
||||
if err := rows.Scan(&schema, &rel, &s.ins, &s.upd, &s.del); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stats[pgx.Identifier{schema, rel}.Sanitize()] = s
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(stats) == 0 {
|
||||
t.Fatal("no user tables")
|
||||
}
|
||||
for table, s := range stats {
|
||||
if err := conn.QueryRow(ctx, "SELECT count(*) FROM "+table).Scan(&s.rows); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stats[table] = s
|
||||
}
|
||||
return stats
|
||||
}
|
||||
|
||||
// lockedBuffer collects a child's output from two pipes.
|
||||
type lockedBuffer struct {
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *lockedBuffer) Write(p []byte) (int, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.Write(p)
|
||||
}
|
||||
|
||||
func (b *lockedBuffer) String() string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.String()
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func TestBootModeReadOnly(t *testing.T) {
|
||||
if got, want := bootModeFor(false), (bootMode{true, true, true, true, true, true, true}); got != want {
|
||||
t.Errorf("bootModeFor(false) = %+v, want %+v", got, want)
|
||||
}
|
||||
if got := bootModeFor(true); got != (bootMode{}) {
|
||||
t.Errorf("bootModeFor(true) = %+v, want every step off", got)
|
||||
}
|
||||
|
||||
cfg, err := pgxpool.ParseConfig("postgres://beacon@127.0.0.1:5432/beacon?application_name=beacon-maint")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
readOnlyTune("maint", cfg)
|
||||
if got := cfg.ConnConfig.RuntimeParams["default_transaction_read_only"]; got != "on" {
|
||||
t.Errorf("default_transaction_read_only = %q, want on", got)
|
||||
}
|
||||
if got := cfg.ConnConfig.RuntimeParams["application_name"]; got != "beacon-maint" {
|
||||
t.Errorf("application_name = %q, want it kept", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,613 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
// Command planguard checks that the planner keeps choosing the plans Beacon's statements were designed for.
|
||||
//
|
||||
// planguard run --tier fixture|local|prod [--corpus db/planguard/corpus.yaml] [--dsn $PLANGUARD_DSN]
|
||||
// [--out planguard-out] [--only RE] [--skip RE] [--modes m,..] [--include-pending] [--search-path P]
|
||||
// [--baseline check|write|off] [--emit resolvers|plans --params FILE] [--plans FILE --params FILE]
|
||||
// [--fidelity-against results.jsonl]
|
||||
// planguard gen-policy [--corpus FILE] [--out FILE] (defaults and db/sqlc are under the module root)
|
||||
// planguard diff --cases db/planguard/diff/NAME.yaml [--old-ref origin/dev] [--walk 1,2,3,web] [--dsn $PLANGUARD_DSN]
|
||||
// [--out planguard-out] [--files db/planguard]
|
||||
// planguard storediff --old URL --new URL --cases db/planguard/storediff/NAME.yaml [--walk 1,2,3,web] [--out DIR]
|
||||
//
|
||||
// diff runs each batch of a case file in one READ ONLY snapshot and compares every pair's full results. old:
|
||||
// statements are read from db/sqlc at --old-ref through git (the checkout is never switched), new: from the working
|
||||
// tree's db/sqlc, file: from --files.
|
||||
//
|
||||
// storediff walks each case of a case file page by page on two running servers (normally both read-only on one
|
||||
// database) and compares the whole JSON of every page.
|
||||
//
|
||||
// --tier fixture first builds the synthetic fixture on --dsn through the migrations (or finds it built from the same
|
||||
// files); it refuses a database that holds data and was not built that way, and one whose collation is not C
|
||||
// (create it with CREATE DATABASE <name> TEMPLATE template0 LC_COLLATE 'C' LC_CTYPE 'C' ENCODING 'UTF8').
|
||||
//
|
||||
// Exit status: 0 when no run fails, 1 when one does (or the run cannot be made), 2 on usage errors.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/db/planpolicy"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/planguard"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
const usage = "usage: planguard run --tier fixture|local|prod [flags] (planguard run -h for the flags)\n" +
|
||||
" planguard gen-policy [--corpus FILE] [--out FILE] (paths default to the module root's db/)\n" +
|
||||
" planguard diff --cases FILE [--old-ref REF] [--walk 1,2,3,web] [--dsn DSN] [--out DIR] [--files DIR]\n" +
|
||||
" planguard storediff --old URL --new URL --cases FILE [--walk 1,2,3,web] [--out DIR]"
|
||||
|
||||
func main() {
|
||||
os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
|
||||
}
|
||||
|
||||
func run(args []string, stdout, stderr io.Writer) int {
|
||||
if len(args) == 0 {
|
||||
fmt.Fprintln(stderr, usage)
|
||||
return 2
|
||||
}
|
||||
switch args[0] {
|
||||
case "run":
|
||||
return runCmd(args[1:], stdout, stderr)
|
||||
case "gen-policy":
|
||||
return genPolicyCmd(args[1:], stdout, stderr)
|
||||
case "diff":
|
||||
return diffCmd(args[1:], stderr)
|
||||
case "storediff":
|
||||
return storeDiffCmd(args[1:], stderr)
|
||||
case "-h", "-help", "--help", "help":
|
||||
fmt.Fprintln(stderr, usage)
|
||||
return 2
|
||||
}
|
||||
fmt.Fprintf(stderr, "planguard: unknown subcommand %q\n%s\n", args[0], usage)
|
||||
return 2
|
||||
}
|
||||
|
||||
// genPolicyCmd writes db/planpolicy/cached_gen.go from the corpus: the cached statements whose entries prove
|
||||
// every shape and mode of their admission profile. Statement texts come from db/sqlc of the module that holds the
|
||||
// working directory (found by walking up to go.mod), so it runs from any directory of the checkout.
|
||||
func genPolicyCmd(args []string, stdout, stderr io.Writer) int {
|
||||
fs := flag.NewFlagSet("gen-policy", flag.ContinueOnError)
|
||||
fs.SetOutput(stderr)
|
||||
corpus := fs.String("corpus", "", "corpus file (default <module root>/db/planguard/corpus.yaml)")
|
||||
out := fs.String("out", "", "generated Go file (default <module root>/db/planpolicy/cached_gen.go)")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return 2
|
||||
}
|
||||
if fs.NArg() > 0 {
|
||||
fmt.Fprintf(stderr, "planguard gen-policy: unexpected argument %q\n", fs.Arg(0))
|
||||
return 2
|
||||
}
|
||||
root, err := moduleRoot()
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard gen-policy:", err)
|
||||
return 1
|
||||
}
|
||||
if *corpus == "" {
|
||||
*corpus = filepath.Join(root, "db", "planguard", "corpus.yaml")
|
||||
}
|
||||
if *out == "" {
|
||||
*out = filepath.Join(root, "db", "planpolicy", "cached_gen.go")
|
||||
}
|
||||
sqlc, err := planguard.LoadStatements(filepath.Join(root, "db", "sqlc"))
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
c, err := planguard.LoadCorpus(os.DirFS(filepath.Dir(*corpus)), filepath.Base(*corpus), sqlc)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
src, err := planguard.GenPolicy(c, planpolicy.Legacy)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard gen-policy:", err)
|
||||
return 1
|
||||
}
|
||||
if old, err := os.ReadFile(*out); err == nil && string(old) == string(src) {
|
||||
fmt.Fprintf(stdout, "%s is up to date\n", *out)
|
||||
return 0
|
||||
}
|
||||
if err := os.WriteFile(*out, src, 0o644); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
fmt.Fprintf(stdout, "wrote %s\n", *out)
|
||||
return 0
|
||||
}
|
||||
|
||||
// diffCmd compares old and new statement results on one snapshot per batch.
|
||||
func diffCmd(args []string, stderr io.Writer) int {
|
||||
fs := flag.NewFlagSet("diff", flag.ContinueOnError)
|
||||
fs.SetOutput(stderr)
|
||||
cases := fs.String("cases", "", "diff case file, e.g. db/planguard/diff/planted_readmodels.yaml (required)")
|
||||
oldRef := fs.String("old-ref", "", "git ref whose db/sqlc gives the old: statements (required when the cases use old:)")
|
||||
walk := fs.String("walk", "", "comma list of page sizes to walk the pairs that have a walk at; web = each walk's web_limit")
|
||||
dsn := fs.String("dsn", os.Getenv("PLANGUARD_DSN"), "PostgreSQL DSN (default $PLANGUARD_DSN)")
|
||||
out := fs.String("out", "planguard-out", "output directory")
|
||||
files := fs.String("files", "", "directory file: statements are relative to (default <module root>/db/planguard)")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return 2
|
||||
}
|
||||
usageErr := func(format string, a ...any) int {
|
||||
fmt.Fprintf(stderr, "planguard diff: "+format+"\n", a...)
|
||||
return 2
|
||||
}
|
||||
if fs.NArg() > 0 {
|
||||
return usageErr("unexpected argument %q", fs.Arg(0))
|
||||
}
|
||||
if *cases == "" {
|
||||
return usageErr("--cases is required")
|
||||
}
|
||||
limits, err := parseWalk(*walk)
|
||||
if err != nil {
|
||||
return usageErr("--walk: %v", err)
|
||||
}
|
||||
if *dsn == "" {
|
||||
return usageErr("--dsn (or $PLANGUARD_DSN) is required")
|
||||
}
|
||||
root, err := moduleRoot()
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff:", err)
|
||||
return 1
|
||||
}
|
||||
if *files == "" {
|
||||
*files = filepath.Join(root, "db", "planguard")
|
||||
}
|
||||
raw, err := os.ReadFile(*cases)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff:", err)
|
||||
return 1
|
||||
}
|
||||
fsys := os.DirFS(*files)
|
||||
spec, err := planguard.ParseDiffSpec(raw, *cases, fsys)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff:", err)
|
||||
return 1
|
||||
}
|
||||
needOld := false
|
||||
for _, b := range spec.Batches {
|
||||
for _, src := range b.Statements {
|
||||
needOld = needOld || strings.HasPrefix(src, "old:")
|
||||
}
|
||||
}
|
||||
var oldSQL map[string]string
|
||||
if needOld {
|
||||
if *oldRef == "" {
|
||||
return usageErr("--old-ref is required: %s uses old: statements", *cases)
|
||||
}
|
||||
if oldSQL, err = planguard.LoadStatementsAtRef(root, *oldRef); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff:", err)
|
||||
return 1
|
||||
}
|
||||
}
|
||||
newSQL, err := planguard.LoadStatements(filepath.Join(root, "db", "sqlc"))
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff:", err)
|
||||
return 1
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
pc, err := pgconn.Connect(ctx, *dsn)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff: connect:", err)
|
||||
return 1
|
||||
}
|
||||
defer pc.Close(context.Background())
|
||||
rep, runErr := planguard.RunDiff(ctx, planguard.NewSession(pc), spec, oldSQL, newSQL, fsys, *out, limits)
|
||||
code := 0
|
||||
if rep != nil {
|
||||
for _, b := range rep.Batches {
|
||||
for _, p := range b.Pairs {
|
||||
fmt.Fprintln(stderr, p.Line())
|
||||
}
|
||||
}
|
||||
fmt.Fprintf(stderr, "report: %s\n", filepath.Join(*out, "diff_report.md"))
|
||||
if n := len(rep.Failed()); n > 0 {
|
||||
fmt.Fprintf(stderr, "%d pairs missed their expectation\n", n)
|
||||
code = 1
|
||||
}
|
||||
if n := len(rep.Skipped()); n > 0 {
|
||||
fmt.Fprintf(stderr, "%d pairs were skipped: their {sql: …} params found no value on this snapshot, so they prove nothing\n", n)
|
||||
code = 1
|
||||
}
|
||||
}
|
||||
if runErr != nil {
|
||||
fmt.Fprintln(stderr, "planguard diff:", runErr)
|
||||
code = 1
|
||||
}
|
||||
return code
|
||||
}
|
||||
|
||||
// storeDiffCmd compares the pages of two servers' API.
|
||||
func storeDiffCmd(args []string, stderr io.Writer) int {
|
||||
fs := flag.NewFlagSet("storediff", flag.ContinueOnError)
|
||||
fs.SetOutput(stderr)
|
||||
oldBase := fs.String("old", "", "base URL of the old server, e.g. http://127.0.0.1:8481 (required)")
|
||||
newBase := fs.String("new", "", "base URL of the new server, e.g. http://127.0.0.1:8482 (required)")
|
||||
cases := fs.String("cases", "", "storediff case file, e.g. db/planguard/storediff/smoke.yaml (required)")
|
||||
walk := fs.String("walk", "web", "comma list of page sizes; web = each case's web_limit")
|
||||
out := fs.String("out", "", "directory for storediff_report.md (default: no file)")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return 2
|
||||
}
|
||||
usageErr := func(format string, a ...any) int {
|
||||
fmt.Fprintf(stderr, "planguard storediff: "+format+"\n", a...)
|
||||
return 2
|
||||
}
|
||||
if fs.NArg() > 0 {
|
||||
return usageErr("unexpected argument %q", fs.Arg(0))
|
||||
}
|
||||
if *oldBase == "" || *newBase == "" || *cases == "" {
|
||||
return usageErr("--old, --new and --cases are required")
|
||||
}
|
||||
for _, base := range []string{*oldBase, *newBase} {
|
||||
if u, err := url.Parse(base); err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
|
||||
return usageErr("%q is not an http(s) base URL", base)
|
||||
}
|
||||
}
|
||||
if _, err := parseWalk(*walk); err != nil {
|
||||
return usageErr("--walk: %v", err)
|
||||
}
|
||||
spec, err := planguard.LoadStoreDiffSpec(*cases)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard storediff:", err)
|
||||
return 1
|
||||
}
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
hc := &http.Client{Timeout: 2 * time.Minute}
|
||||
rep, runErr := planguard.RunStoreDiff(ctx, hc, *oldBase, *newBase, spec, strings.Split(*walk, ","))
|
||||
code := 0
|
||||
if rep != nil {
|
||||
for _, c := range rep.Cases {
|
||||
fmt.Fprintln(stderr, c.Line())
|
||||
}
|
||||
if *out != "" {
|
||||
path := filepath.Join(*out, "storediff_report.md")
|
||||
if err := os.MkdirAll(*out, 0o755); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard storediff:", err)
|
||||
code = 1
|
||||
} else if err := os.WriteFile(path, []byte(rep.Markdown()), 0o644); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard storediff:", err)
|
||||
code = 1
|
||||
} else {
|
||||
fmt.Fprintf(stderr, "report: %s\n", path)
|
||||
}
|
||||
}
|
||||
if n := len(rep.Failed()); n > 0 {
|
||||
fmt.Fprintf(stderr, "%d results differ\n", n)
|
||||
code = 1
|
||||
}
|
||||
if n := len(rep.Skipped()); n > 0 {
|
||||
fmt.Fprintf(stderr, "%d results were skipped: a path var found no value, so they prove nothing\n", n)
|
||||
code = 1
|
||||
}
|
||||
}
|
||||
if runErr != nil {
|
||||
fmt.Fprintln(stderr, "planguard storediff:", runErr)
|
||||
code = 1
|
||||
}
|
||||
return code
|
||||
}
|
||||
|
||||
// parseWalk reads "1,2,3,web" (web → 0, each walk's web_limit).
|
||||
func parseWalk(s string) ([]int, error) {
|
||||
if s == "" {
|
||||
return nil, nil
|
||||
}
|
||||
var out []int
|
||||
for _, part := range strings.Split(s, ",") {
|
||||
part = strings.TrimSpace(part)
|
||||
if part == "web" {
|
||||
out = append(out, 0)
|
||||
continue
|
||||
}
|
||||
n, err := strconv.Atoi(part)
|
||||
if err != nil || n < 1 {
|
||||
return nil, fmt.Errorf("%q is not a page size ≥ 1 or web", part)
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// moduleRoot returns the nearest directory at or above the working directory that holds go.mod.
|
||||
func moduleRoot() (string, error) {
|
||||
dir, err := os.Getwd()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
for {
|
||||
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
|
||||
return dir, nil
|
||||
}
|
||||
parent := filepath.Dir(dir)
|
||||
if parent == dir {
|
||||
return "", errors.New("no go.mod at or above the working directory; run it inside the beacon-server checkout")
|
||||
}
|
||||
dir = parent
|
||||
}
|
||||
}
|
||||
|
||||
type runFlags struct {
|
||||
tier, corpus, dsn, out, only, skip, modes, searchPath, baseline, emit, params, plans, fidelity string
|
||||
includePending bool
|
||||
}
|
||||
|
||||
func usageErr(stderr io.Writer, format string, a ...any) int {
|
||||
fmt.Fprintf(stderr, "planguard run: "+format+"\n", a...)
|
||||
return 2
|
||||
}
|
||||
|
||||
func runCmd(args []string, stdout, stderr io.Writer) int {
|
||||
var f runFlags
|
||||
fs := flag.NewFlagSet("run", flag.ContinueOnError)
|
||||
fs.SetOutput(stderr)
|
||||
fs.StringVar(&f.tier, "tier", "", "fixture | local | prod (required)")
|
||||
fs.StringVar(&f.corpus, "corpus", "db/planguard/corpus.yaml", "corpus file; baselines live in its baseline/<tier> directory")
|
||||
fs.StringVar(&f.dsn, "dsn", os.Getenv("PLANGUARD_DSN"), "PostgreSQL DSN (default $PLANGUARD_DSN)")
|
||||
fs.StringVar(&f.out, "out", "planguard-out", "output directory")
|
||||
fs.StringVar(&f.only, "only", "", "regexp: run only entries whose id matches")
|
||||
fs.StringVar(&f.skip, "skip", "", "regexp: leave out entries whose id matches")
|
||||
fs.StringVar(&f.modes, "modes", "", "comma list of plan modes overriding every entry's")
|
||||
fs.BoolVar(&f.includePending, "include-pending", false, "run pending statements too (local overlays only)")
|
||||
fs.StringVar(&f.searchPath, "search-path", "", "override every group's search_path")
|
||||
fs.StringVar(&f.baseline, "baseline", "", "check | write | off (default check; prod is always off)")
|
||||
fs.StringVar(&f.emit, "emit", "", "resolvers | plans: write a read-only psql script to stdout instead of running")
|
||||
fs.StringVar(&f.params, "params", "", "psql output of the --emit resolvers script")
|
||||
fs.StringVar(&f.plans, "plans", "", "psql output of the --emit plans script: check it instead of running")
|
||||
fs.StringVar(&f.fidelity, "fidelity-against", "", "results.<tier>.jsonl of another run to compare drivers with")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return 2
|
||||
}
|
||||
if fs.NArg() > 0 {
|
||||
return usageErr(stderr, "unexpected argument %q", fs.Arg(0))
|
||||
}
|
||||
o, code := f.options(stderr)
|
||||
if code != 0 {
|
||||
return code
|
||||
}
|
||||
|
||||
sqlc, err := planguard.LoadStatements("db/sqlc")
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
c, err := planguard.LoadCorpus(os.DirFS(filepath.Dir(f.corpus)), filepath.Base(f.corpus), sqlc)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
if f.params != "" {
|
||||
if o.Params, err = readResolverOutput(f.params); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
}
|
||||
|
||||
switch {
|
||||
case f.emit == "resolvers":
|
||||
return emit(stderr, func(nonce string) error { return planguard.EmitResolvers(stdout, c, o, nonce) })
|
||||
case f.emit == "plans":
|
||||
return emit(stderr, func(nonce string) error { return planguard.EmitPlans(stdout, c, o, o.Params, nonce) })
|
||||
case f.plans != "":
|
||||
raw, err := os.ReadFile(f.plans)
|
||||
if err == nil {
|
||||
var nonce string
|
||||
if nonce, err = planguard.ScriptNonce(raw); err == nil {
|
||||
var plans map[string][]byte
|
||||
if plans, err = planguard.ParsePlanOutput(strings.NewReader(string(raw)), nonce); err == nil {
|
||||
var rep *planguard.Report
|
||||
rep, err = planguard.CheckPlans(c, o, plans)
|
||||
if err == nil {
|
||||
for _, r := range rep.Results {
|
||||
fmt.Fprintln(stderr, r.Line())
|
||||
}
|
||||
return finish(rep, nil, f, stderr)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
fmt.Fprintf(stderr, "planguard: %s: %v\n", f.plans, err)
|
||||
return 1
|
||||
}
|
||||
|
||||
if f.dsn == "" {
|
||||
return usageErr(stderr, "--dsn (or $PLANGUARD_DSN) is required to run")
|
||||
}
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if f.tier == "fixture" {
|
||||
if err := buildFixture(ctx, f, stderr); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
}
|
||||
pc, err := pgconn.Connect(ctx, f.dsn)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard: connect:", err)
|
||||
return 1
|
||||
}
|
||||
defer pc.Close(context.Background())
|
||||
o.Progress = func(r planguard.Result) { fmt.Fprintln(stderr, r.Line()) }
|
||||
rep, err := planguard.Run(ctx, c, planguard.NewSession(pc), o)
|
||||
if rep == nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
return finish(rep, err, f, stderr)
|
||||
}
|
||||
|
||||
// buildFixture builds the fixture tier's database on --dsn from the fixture/ directory beside the corpus, or finds
|
||||
// it built from the same files.
|
||||
func buildFixture(ctx context.Context, f runFlags, stderr io.Writer) error {
|
||||
pool, err := pgxpool.New(ctx, f.dsn)
|
||||
if err != nil {
|
||||
return fmt.Errorf("connect: %w", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
t0 := time.Now()
|
||||
if err := planguard.BuildFixture(ctx, pool, os.DirFS(filepath.Dir(f.corpus))); err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Fprintf(stderr, "fixture ready in %.1f s\n", time.Since(t0).Seconds())
|
||||
return nil
|
||||
}
|
||||
|
||||
// options checks the flags and builds the run options; code is 2 on a usage error.
|
||||
func (f runFlags) options(stderr io.Writer) (planguard.RunOptions, int) {
|
||||
o := planguard.RunOptions{Tier: f.tier, IncludePending: f.includePending, SearchPath: f.searchPath}
|
||||
switch f.tier {
|
||||
case "":
|
||||
return o, usageErr(stderr, "--tier is required (fixture, local or prod)")
|
||||
case "fixture", "local", "prod":
|
||||
default:
|
||||
return o, usageErr(stderr, "unknown --tier %q (want fixture, local or prod)", f.tier)
|
||||
}
|
||||
switch f.baseline {
|
||||
case "":
|
||||
o.Baseline = planguard.BaselineCheck
|
||||
case "check", "write", "off":
|
||||
o.Baseline = planguard.BaselineMode(f.baseline)
|
||||
default:
|
||||
return o, usageErr(stderr, "unknown --baseline %q (want check, write or off)", f.baseline)
|
||||
}
|
||||
if f.tier == "prod" {
|
||||
o.Baseline = planguard.BaselineOff
|
||||
}
|
||||
o.BaselineDir = filepath.Join(filepath.Dir(f.corpus), "baseline", f.tier)
|
||||
var err error
|
||||
if f.only != "" {
|
||||
if o.Only, err = regexp.Compile(f.only); err != nil {
|
||||
return o, usageErr(stderr, "--only: %v", err)
|
||||
}
|
||||
}
|
||||
if f.skip != "" {
|
||||
if o.Skip, err = regexp.Compile(f.skip); err != nil {
|
||||
return o, usageErr(stderr, "--skip: %v", err)
|
||||
}
|
||||
}
|
||||
if f.modes != "" {
|
||||
for _, m := range strings.Split(f.modes, ",") {
|
||||
switch m = strings.TrimSpace(m); m {
|
||||
case "custom", "generic", "auto":
|
||||
o.Modes = append(o.Modes, m)
|
||||
default:
|
||||
return o, usageErr(stderr, "--modes: unknown mode %q (want custom, generic or auto)", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
if (f.emit != "" || f.plans != "") && f.tier != "prod" {
|
||||
return o, usageErr(stderr, "--emit and --plans are the prod tier's psql path; use them with --tier prod")
|
||||
}
|
||||
switch f.emit {
|
||||
case "", "resolvers":
|
||||
case "plans":
|
||||
if f.params == "" {
|
||||
return o, usageErr(stderr, "--emit plans needs --params (the psql output of --emit resolvers)")
|
||||
}
|
||||
default:
|
||||
return o, usageErr(stderr, "unknown --emit %q (want resolvers or plans)", f.emit)
|
||||
}
|
||||
if f.plans != "" {
|
||||
if f.emit != "" {
|
||||
return o, usageErr(stderr, "--plans and --emit are exclusive")
|
||||
}
|
||||
if f.params == "" {
|
||||
return o, usageErr(stderr, "--plans needs --params (the psql output of --emit resolvers)")
|
||||
}
|
||||
}
|
||||
return o, 0
|
||||
}
|
||||
|
||||
func emit(stderr io.Writer, write func(nonce string) error) int {
|
||||
nonce, err := planguard.NewNonce()
|
||||
if err == nil {
|
||||
err = write(nonce)
|
||||
}
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard:", err)
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func readResolverOutput(path string) (map[string]*string, error) {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
nonce, err := planguard.ScriptNonce(raw)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", path, err)
|
||||
}
|
||||
res, err := planguard.ParseResolverOutput(strings.NewReader(string(raw)), nonce)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", path, err)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// finish writes the report files, prints the totals and the fidelity, and returns the exit status.
|
||||
func finish(rep *planguard.Report, runErr error, f runFlags, stderr io.Writer) int {
|
||||
code := 0
|
||||
if err := rep.WriteFiles(f.out); err != nil {
|
||||
fmt.Fprintln(stderr, "planguard: write report:", err)
|
||||
code = 1
|
||||
}
|
||||
fmt.Fprintf(stderr, "%d runs in %.1f s: %s\n", len(rep.Results), rep.DurationMs/1000, rep.CountsString())
|
||||
if f.fidelity != "" {
|
||||
other, err := planguard.ReadResults(f.fidelity)
|
||||
if err != nil {
|
||||
fmt.Fprintln(stderr, "planguard: fidelity:", err)
|
||||
code = 1
|
||||
} else {
|
||||
var otherIx []string
|
||||
if meta, err := planguard.ReadReportMeta(f.fidelity); err == nil {
|
||||
otherIx = meta.Indexes
|
||||
} else if errors.Is(err, os.ErrNotExist) {
|
||||
fmt.Fprintf(stderr, "planguard: fidelity: no report file beside %s; its index set is unknown, so nothing is excluded for it\n", f.fidelity)
|
||||
} else {
|
||||
fmt.Fprintln(stderr, "planguard: fidelity:", err)
|
||||
code = 1
|
||||
}
|
||||
fr := planguard.Fidelity(rep.Results, other, rep.Indexes, otherIx)
|
||||
pct := 0.0
|
||||
if fr.Total > 0 {
|
||||
pct = 100 * float64(fr.Match) / float64(fr.Total)
|
||||
}
|
||||
fmt.Fprintf(stderr, "fidelity: %d of %d driver sets match (%.1f %%), %d excluded for index-set differences\n", fr.Match, fr.Total, pct, fr.Excluded)
|
||||
for _, m := range fr.Mismatches {
|
||||
fmt.Fprintln(stderr, " differs:", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
if runErr != nil {
|
||||
fmt.Fprintln(stderr, "planguard: run stopped:", runErr)
|
||||
code = 1
|
||||
}
|
||||
if n := len(rep.Failures()); n > 0 {
|
||||
fmt.Fprintf(stderr, "%d failing runs\n", n)
|
||||
code = 1
|
||||
}
|
||||
return code
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestCLIUsage(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
args []string
|
||||
want string // in stderr
|
||||
}{
|
||||
{"no args", nil, "usage"},
|
||||
{"unknown subcommand", []string{"frobnicate"}, "frobnicate"},
|
||||
{"run without tier", []string{"run"}, "--tier"},
|
||||
{"run with unknown tier", []string{"run", "--tier", "staging"}, "staging"},
|
||||
{"bad baseline", []string{"run", "--tier", "local", "--baseline", "maybe"}, "--baseline"},
|
||||
{"bad only", []string{"run", "--tier", "local", "--only", "("}, "--only"},
|
||||
{"emit plans without params", []string{"run", "--tier", "prod", "--emit", "plans"}, "--params"},
|
||||
{"emit on local", []string{"run", "--tier", "local", "--emit", "resolvers"}, "--tier prod"},
|
||||
{"plans on fixture", []string{"run", "--tier", "fixture", "--plans", "p.out", "--params", "r.out"}, "--tier prod"},
|
||||
{"unknown flag", []string{"run", "--tier", "local", "--nope"}, "nope"},
|
||||
{"gen-policy unknown flag", []string{"gen-policy", "--nope"}, "nope"},
|
||||
{"gen-policy argument", []string{"gen-policy", "extra"}, "extra"},
|
||||
{"diff without cases", []string{"diff", "--dsn", "x"}, "--cases"},
|
||||
{"diff bad walk", []string{"diff", "--cases", "c.yaml", "--walk", "1,0"}, "--walk"},
|
||||
{"diff bad walk word", []string{"diff", "--cases", "c.yaml", "--walk", "web,all"}, "all"},
|
||||
{"diff unknown flag", []string{"diff", "--nope"}, "nope"},
|
||||
{"storediff without new", []string{"storediff", "--old", "http://a", "--cases", "c.yaml"}, "--new"},
|
||||
{"storediff bad url", []string{"storediff", "--old", "a:8481", "--new", "http://b", "--cases", "c.yaml"}, "a:8481"},
|
||||
{"storediff bad walk", []string{"storediff", "--old", "http://a", "--new", "http://b", "--cases", "c.yaml", "--walk", "0"}, "--walk"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var out, errb bytes.Buffer
|
||||
if code := run(tc.args, &out, &errb); code != 2 {
|
||||
t.Errorf("exit %d, want 2 (stderr %q)", code, errb.String())
|
||||
}
|
||||
if !strings.Contains(errb.String(), tc.want) {
|
||||
t.Errorf("stderr %q, want it to name %q", errb.String(), tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// gen-policy finds the module root from a subdirectory. It writes to a temporary file, which must equal the
|
||||
// committed cached_gen.go; the committed file is never written.
|
||||
func TestGenPolicyFromSubdirectory(t *testing.T) {
|
||||
want, err := os.ReadFile("../../db/planpolicy/cached_gen.go")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out := filepath.Join(t.TempDir(), "cached_gen.go")
|
||||
t.Chdir("../../internal")
|
||||
var stdout, errb bytes.Buffer
|
||||
if code := run([]string{"gen-policy", "--out", out}, &stdout, &errb); code != 0 {
|
||||
t.Fatalf("exit %d: %s", code, errb.String())
|
||||
}
|
||||
got, err := os.ReadFile(out)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("generated policy differs from the committed db/planpolicy/cached_gen.go:\n%s", got)
|
||||
}
|
||||
}
|
||||
|
||||
// storediff exits 0 when every page matches and 1 when one differs, and writes its report under --out.
|
||||
func TestStoreDiffCLI(t *testing.T) {
|
||||
serve := func(name string) *httptest.Server {
|
||||
s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprintf(w, `{"items":[{"id":1,"name":%q}],"hasMore":false}`, name)
|
||||
}))
|
||||
t.Cleanup(s.Close)
|
||||
return s
|
||||
}
|
||||
dir := t.TempDir()
|
||||
cases := filepath.Join(dir, "cases.yaml")
|
||||
if err := os.WriteFile(cases, []byte("cases:\n - id: things\n path: /api/v1/things\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
a, b, c := serve("x"), serve("x"), serve("y")
|
||||
var out, errb bytes.Buffer
|
||||
if code := run([]string{"storediff", "--old", a.URL, "--new", b.URL, "--cases", cases, "--walk", "1,web", "--out", dir}, &out, &errb); code != 0 {
|
||||
t.Fatalf("same servers: exit %d, stderr %s", code, errb.String())
|
||||
}
|
||||
report, err := os.ReadFile(filepath.Join(dir, "storediff_report.md"))
|
||||
if err != nil || !strings.Contains(string(report), "| things | 50 | 1 | identical |") {
|
||||
t.Errorf("report %q (%v)", report, err)
|
||||
}
|
||||
errb.Reset()
|
||||
if code := run([]string{"storediff", "--old", a.URL, "--new", c.URL, "--cases", cases}, &out, &errb); code != 1 ||
|
||||
!strings.Contains(errb.String(), `$.items[0].name: old "x", new "y"`) {
|
||||
t.Errorf("different servers: exit %d, stderr %s", code, errb.String())
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,9 @@ server:
|
||||
# Address of the admin listener serving Prometheus /metrics, e.g. 127.0.0.1:9090. Off when empty.
|
||||
# It has no authentication: bind it to a private interface.
|
||||
admin_listen: ""
|
||||
# read_only: true boots a server that never writes: no migrations, no ingest, no background jobs, no
|
||||
# MeshMapper imports, and every database session is read-only. Used to compare two builds on one database.
|
||||
read_only: false
|
||||
|
||||
# Applies to /api/v1/* only. Both counters are local to this Beacon instance.
|
||||
# Set trusted_proxies correctly before using a reverse proxy, or clients share
|
||||
@@ -69,6 +72,11 @@ ratelimit:
|
||||
# enforce: []
|
||||
# budget: shadow # the per-client DB-time budget only records; enforcement is not available
|
||||
|
||||
# Daily read-only plan regression check (GET /api/v1/admin/db/planguard, beacon_planguard_* metrics).
|
||||
# Enable on large deployments; plan expectations assume production-sized tables.
|
||||
#planguard:
|
||||
# daily_check: false
|
||||
|
||||
# MeshMapper integrations (optional, require a regional or grouped-region API key).
|
||||
# Your local MeshMapper regional or grouped-region admin can generate the key.
|
||||
# A grouped-region key covers multiple regions.
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
[
|
||||
"Aggregate partial=Simple",
|
||||
" Limit [InitPlan]",
|
||||
" LockRows",
|
||||
" Incremental Sort",
|
||||
" Index Scan on packets vp using idx_packets_last_heard scan=Backward",
|
||||
" ModifyTable on packets dp [InitPlan]",
|
||||
" Nested Loop",
|
||||
" CTE Scan",
|
||||
" Index Scan on packets dp using packets_pkey scan=Forward",
|
||||
" ModifyTable on analytics_raw_state [InitPlan]",
|
||||
" Aggregate partial=Simple [InitPlan]",
|
||||
" CTE Scan",
|
||||
" CTE Scan [InitPlan]",
|
||||
" Result",
|
||||
" Seq Scan on analytics_raw_state",
|
||||
" CTE Scan"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Incremental Sort",
|
||||
" Index Scan on packets vp using idx_packets_last_heard scan=Backward"
|
||||
]
|
||||
@@ -0,0 +1,18 @@
|
||||
[
|
||||
"Result",
|
||||
" Limit [InitPlan]",
|
||||
" LockRows",
|
||||
" Index Scan on known_routes r using idx_known_routes_keyset scan=Backward",
|
||||
" ModifyTable on known_routes kr [InitPlan]",
|
||||
" Nested Loop",
|
||||
" CTE Scan",
|
||||
" Index Scan on known_routes kr using known_routes_pkey scan=Forward",
|
||||
" Aggregate partial=Simple [InitPlan]",
|
||||
" CTE Scan",
|
||||
" Limit [InitPlan]",
|
||||
" Sort",
|
||||
" CTE Scan",
|
||||
" Limit [InitPlan]",
|
||||
" Sort",
|
||||
" CTE Scan"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes r using idx_known_routes_keyset scan=Backward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes r using idx_known_routes_keyset scan=Backward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes r using idx_known_routes_keyset scan=Backward"
|
||||
]
|
||||
@@ -0,0 +1,28 @@
|
||||
[
|
||||
"Result",
|
||||
" Limit [InitPlan]",
|
||||
" Index Scan on known_routes r using idx_known_routes_keyset scan=Backward",
|
||||
" LockRows [InitPlan]",
|
||||
" Nested Loop",
|
||||
" CTE Scan",
|
||||
" Index Scan on known_routes r_1 using known_routes_pkey scan=Forward",
|
||||
" ModifyTable on known_routes kr [InitPlan]",
|
||||
" Nested Loop",
|
||||
" CTE Scan",
|
||||
" Index Scan on known_routes kr using known_routes_pkey scan=Forward",
|
||||
" Aggregate partial=Simple [InitPlan]",
|
||||
" CTE Scan",
|
||||
" Aggregate partial=Simple [InitPlan]",
|
||||
" CTE Scan",
|
||||
" Limit [InitPlan]",
|
||||
" Sort",
|
||||
" CTE Scan",
|
||||
" Limit [InitPlan]",
|
||||
" Sort",
|
||||
" CTE Scan",
|
||||
" Aggregate partial=Simple [InitPlan]",
|
||||
" Hash Join (Anti)",
|
||||
" CTE Scan",
|
||||
" Hash",
|
||||
" CTE Scan"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes r using idx_known_routes_keyset scan=Backward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_iata_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,6 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" BitmapAnd",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids",
|
||||
" Bitmap Index Scan using known_routes_pkey"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,5 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Index Scan on known_routes using idx_known_routes_hop_iata scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,4 @@
|
||||
[
|
||||
"Limit",
|
||||
" Index Scan on known_routes using idx_known_routes_keyset scan=Forward"
|
||||
]
|
||||
@@ -0,0 +1,6 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids"
|
||||
]
|
||||
@@ -0,0 +1,6 @@
|
||||
[
|
||||
"Limit",
|
||||
" Sort",
|
||||
" Bitmap Heap Scan on known_routes kr",
|
||||
" Bitmap Index Scan using idx_known_routes_node_ids"
|
||||
]
|
||||
Loaded 100 of 327 files, more files were not shown because too many files have changed in this diff.
Show more
Reference in new issue
Block a user