diff --git a/internal/query/batch_cache.go b/internal/query/batch_cache.go index 139668cf..a1a085c6 100644 --- a/internal/query/batch_cache.go +++ b/internal/query/batch_cache.go @@ -17,7 +17,7 @@ import ( // readCacheVersion covers both schema and aggregation semantics. These tables // are disposable; a mismatch rebuilds them together from immutable Parquet. -const readCacheVersion = 4 +const readCacheVersion = 5 const ( readCacheBatchLimit = 64 @@ -62,6 +62,12 @@ func createBatchCaches(db *sql.DB) error { return err } } + // Bucket counts and the unfinished edge window are one disposable + // unit. Restart that window from its unchanged ingestion watermark + // when upgrading from a cache without these counts (or new semantics). + if _, err := tx.Exec(`DELETE FROM rollup_state WHERE cache_key IN (?, ?)`, edgeRollupSubCursorKey, edgeRollupSubWindowKey); err != nil { + return err + } if _, err := tx.Exec(`DELETE FROM read_cache_version`); err != nil { return err } @@ -70,6 +76,7 @@ func createBatchCaches(db *sql.DB) error { } } for _, stmt := range []string{ + `CREATE TABLE IF NOT EXISTS read_edge_buckets (namespace VARCHAR, bucket TIMESTAMPTZ_NS, spans BIGINT)`, `CREATE TABLE IF NOT EXISTS read_batches (batch_id VARCHAR PRIMARY KEY,replacement_id VARCHAR)`, `CREATE TABLE IF NOT EXISTS read_trace_candidates (trace_id VARCHAR PRIMARY KEY,namespace VARCHAR,service VARCHAR,min_start BIGINT,max_start BIGINT,max_end BIGINT,has_error INTEGER)`, `CREATE TABLE IF NOT EXISTS read_trace_parts (batch_id VARCHAR, namespace VARCHAR, service VARCHAR, trace_id VARCHAR, min_start BIGINT, max_start BIGINT, max_end BIGINT, has_error INTEGER)`, diff --git a/internal/query/duck.go b/internal/query/duck.go index 6820a46f..2465ce9c 100644 --- a/internal/query/duck.go +++ b/internal/query/duck.go @@ -342,16 +342,14 @@ func (d *Duck) skipRollupToLatest(ctx context.Context) error { }{ {serviceRollupStateKey, horizon}, {edgeRollupStateKey, horizon}, - // Skipping discards any half-aggregated window, so its resume point - // must go too or the next pass would reopen a window behind the - // watermark it just advanced. - {edgeRollupSubCursorKey, 0}, - {edgeRollupSubWindowKey, 0}, } { if err := storeRollupWatermark(ctx, tx, w.key, w.watermark); err != nil { return err } } + if err := clearEdgeRollupCursor(ctx, tx); err != nil { + return err + } if err := tx.Commit(); err != nil { return err } @@ -960,12 +958,19 @@ func (d *Duck) refreshEdgeRollupPass(ctx context.Context) (int64, bool, error) { } newWatermark := windowEnd - // Build the incoming/recent-arrival repair relation once in this snapshot. - // Count every affected span per minute so adaptive counting preserves repair - // duplicates without retaining all of their individual rows. - // TEMP is connection-local; creation rolls back on failure, and the table - // is explicitly dropped before commit so no later pass can observe it. - if _, err := tx.ExecContext(ctx, createEdgePassBucketsSQL, windowStart, windowEnd); err != nil { + // Discover affected buckets once per frozen ingestion window, including + // recent-arrival repairs. Persist only their counts, atomically with the + // cursor. Resumes read these small summaries rather than the original spans. + if !resuming { + if _, err := tx.ExecContext(ctx, "DELETE FROM read_edge_buckets"); err != nil { + return 0, false, err + } + if _, err := tx.ExecContext(ctx, "INSERT INTO read_edge_buckets "+edgePassBucketsSelectSQL, windowStart, windowEnd); err != nil { + return 0, false, err + } + } + // Keep the per-pass name connection-local. It never holds individual spans. + if _, err := tx.ExecContext(ctx, "CREATE TEMP VIEW edge_pass_buckets AS SELECT * FROM read_edge_buckets"); err != nil { return 0, false, err } @@ -1037,6 +1042,9 @@ FROM edge_pass_buckets`).Scan(&minStartT, &maxStartT) } } else { storedWatermark = lastWatermark + if _, err := tx.ExecContext(ctx, `DELETE FROM read_edge_buckets WHERE bucket < ?::TIMESTAMP_NS::TIMESTAMPTZ_NS`, time.Unix(0, nextCursor)); err != nil { + return 0, false, err + } if err := storeRollupWatermark(ctx, tx, edgeRollupSubCursorKey, nextCursor); err != nil { return 0, false, err } @@ -1044,7 +1052,7 @@ FROM edge_pass_buckets`).Scan(&minStartT, &maxStartT) return 0, false, err } } - if _, err := tx.ExecContext(ctx, "DROP TABLE edge_pass_buckets"); err != nil { + if _, err := tx.ExecContext(ctx, "DROP VIEW edge_pass_buckets"); err != nil { return 0, false, err } if err := tx.Commit(); err != nil { @@ -1058,9 +1066,12 @@ FROM edge_pass_buckets`).Scan(&minStartT, &maxStartT) return totalAffected, !completed || chunked, nil } -// clearEdgeRollupCursor drops the resume point once its window is fully -// aggregated, so the next pass opens a fresh ingested window. +// clearEdgeRollupCursor discards the resume point and its bucket summaries +// together, so the next pass opens a fresh ingested window. func clearEdgeRollupCursor(ctx context.Context, tx *sql.Tx) error { + if _, err := tx.ExecContext(ctx, "DELETE FROM read_edge_buckets"); err != nil { + return err + } if err := storeRollupWatermark(ctx, tx, edgeRollupSubCursorKey, 0); err != nil { return err } @@ -1413,8 +1424,9 @@ const edgeArrivalGap = 5 * time.Minute // Only relationship keys are read from the recent ingestion range. Its stamp // predicates allow Parquet footer pruning independently of event-time history. // Messaging uses the aggregate's exact keys, including the shared minute. -var createEdgePassBucketsSQL = fmt.Sprintf(` -CREATE TEMP TABLE edge_pass_buckets AS +var createEdgePassBucketsSQL = "CREATE TEMP TABLE edge_pass_buckets AS " + edgePassBucketsSelectSQL + +var edgePassBucketsSelectSQL = fmt.Sprintf(` WITH incoming AS MATERIALIZED ( SELECT namespace, trace_id, span_id, start_time, kind, messaging_destination, messaging_system diff --git a/internal/query/edge_pass_test.go b/internal/query/edge_pass_test.go index 4af10537..0e3658b5 100644 --- a/internal/query/edge_pass_test.go +++ b/internal/query/edge_pass_test.go @@ -110,7 +110,7 @@ func TestEdgePassDropsRelationOnEveryExit(t *testing.T) { d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return stamp, stamp, stamp }} checkAbsent := func() { var count int - if err := db.QueryRow("SELECT count(*) FROM duckdb_tables() WHERE table_name IN ('edge_pass_buckets','edge_pass_parents','edge_pass_unmatched')").Scan(&count); err != nil || count != 0 { + if err := db.QueryRow("SELECT count(*) FROM (SELECT table_name AS name FROM duckdb_tables() UNION ALL SELECT view_name AS name FROM duckdb_views()) WHERE name IN ('edge_pass_buckets','edge_pass_parents','edge_pass_unmatched')").Scan(&count); err != nil || count != 0 { t.Fatalf("pass relation survived: count=%d err=%v", count, err) } } @@ -126,6 +126,10 @@ func TestEdgePassDropsRelationOnEveryExit(t *testing.T) { if watermark, cursor := readEdgeRollupState(t, d); watermark != 0 || cursor != 0 { t.Fatalf("failed pass advanced state: watermark=%d cursor=%d", watermark, cursor) } + var buckets int + if err := db.QueryRow("SELECT count(*) FROM read_edge_buckets").Scan(&buckets); err != nil || buckets != 0 { + t.Fatalf("failed discovery retained buckets=%d err=%v", buckets, err) + } if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN saved_avg_ms TO avg_ms"); err != nil { t.Fatal(err) } @@ -135,6 +139,25 @@ func TestEdgePassDropsRelationOnEveryExit(t *testing.T) { } checkAbsent() if !empty { + watermark, cursor := readEdgeRollupState(t, d) + // A failed resume must retain the previous cursor AND summaries. + if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN avg_ms TO saved_avg_ms"); err != nil { + t.Fatal(err) + } + if _, _, err := d.refreshEdgeRollupPass(t.Context()); err == nil { + t.Fatal("incompatible rollup schema did not fail the resume") + } + checkAbsent() + if w, c := readEdgeRollupState(t, d); w != watermark || c != cursor { + t.Fatalf("failed resume moved state: watermark=%d cursor=%d", w, c) + } + var buckets int + if err := db.QueryRow("SELECT count(*) FROM read_edge_buckets").Scan(&buckets); err != nil || buckets != 1 { + t.Fatalf("failed resume lost summaries: buckets=%d err=%v", buckets, err) + } + if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN saved_avg_ms TO avg_ms"); err != nil { + t.Fatal(err) + } if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || pending { t.Fatalf("resumed pass pending=%v err=%v", pending, err) } diff --git a/internal/query/edge_resume_test.go b/internal/query/edge_resume_test.go new file mode 100644 index 00000000..7026efb5 --- /dev/null +++ b/internal/query/edge_resume_test.go @@ -0,0 +1,140 @@ +package query + +import ( + "database/sql" + "fmt" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestEdgeResumeSetupIsLinearAcrossRestart(t *testing.T) { + for _, minutes := range []int{5, 15, 30} { + t.Run(fmt.Sprintf("%d_minutes", minutes), func(t *testing.T) { + path := filepath.Join(t.TempDir(), "cache.duckdb") + var db *sql.DB + open := func() { + var err error + db, err = sql.Open("duckdb", path) + if err != nil { + t.Fatal(err) + } + db.SetMaxOpenConns(1) + if _, err := db.Exec("SET TimeZone='UTC'"); err != nil { + t.Fatal(err) + } + if err := CreateTables(db); err != nil { + t.Fatal(err) + } + } + open() + defer func() { _ = db.Close() }() + base := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + stamp := base.UnixNano() + // N occupied event minutes, separated to force N resumable passes + // without needing 250k spans per minute in the regular test suite. + if _, err := db.Exec(`INSERT INTO telemetry.spans + (namespace,trace_id,span_id,parent_span_id,service,kind,start_time,duration_ms,status,ingested_unix_nano) + SELECT 'ns',(i//2)::VARCHAR,i::VARCHAR,CASE WHEN i%2=1 THEN (i-1)::VARCHAR END, + CASE WHEN i%2=0 THEN 'gateway' ELSE 'checkout' END,'SPAN_KIND_SERVER', + ?::TIMESTAMP_NS + (i//2)*INTERVAL 31 MINUTE,5,'OK',? FROM range(?) t(i)`, base, stamp, minutes*2); err != nil { + t.Fatal(err) + } + if _, err := db.Exec("CREATE SEQUENCE edge_setup_reads"); err != nil { + t.Fatal(err) + } + // Only discovery uses ingestion stamps. Count actual source-row + // evaluations there, rather than timing the drain or inspecting SQL. + view := strings.Replace(viewSpans, " ingested_unix_nano,", " CASE WHEN nextval('edge_setup_reads')>0 THEN ingested_unix_nano END AS ingested_unix_nano,", 1) + if _, err := db.Exec(view); err != nil { + t.Fatal(err) + } + var setupReads int64 + for pass := 0; pass < minutes; pass++ { + d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return stamp, stamp, stamp }} + _, pending, err := d.refreshEdgeRollupPass(t.Context()) + if err != nil || pending != (pass < minutes-1) { + t.Fatalf("pass=%d pending=%v err=%v", pass, pending, err) + } + var reads int64 + if err := db.QueryRow("SELECT last_value FROM duckdb_sequences() WHERE sequence_name='edge_setup_reads'").Scan(&reads); err != nil { + t.Fatal(err) + } + if pass == 0 { + setupReads = reads + if reads < int64(minutes*2) || reads > int64(minutes*8) { + t.Fatalf("setup reads=%d, want a constant number per input row", reads) + } + } else if reads != setupReads { + t.Fatalf("resume repeated discovery: reads=%d first pass=%d", reads, setupReads) + } + var remaining int + if err := db.QueryRow("SELECT count(*) FROM read_edge_buckets").Scan(&remaining); err != nil || remaining != minutes-pass-1 { + t.Fatalf("remaining=%d pass=%d err=%v", remaining, pass, err) + } + if pass == 0 { + // Close the native database: summaries and cursor must survive, + // while the connection-local pass view must have disappeared. + if err := db.Close(); err != nil { + t.Fatal(err) + } + open() + } + } + var calls, buckets int + if err := db.QueryRow("SELECT sum(calls),count(DISTINCT bucket) FROM edge_rollup").Scan(&calls, &buckets); err != nil || calls != minutes || buckets != minutes { + t.Fatalf("calls=%d buckets=%d err=%v", calls, buckets, err) + } + t.Logf("occupied_minutes=%d setup_row_evaluations=%d passes=%d", minutes, setupReads, minutes) + }) + } +} + +func TestEdgeResumeKeepsFrozenWindowAndRepairsNewArrivals(t *testing.T) { + db := openTestDuck(t) + if err := CreateTables(db); err != nil { + t.Fatal(err) + } + if err := CreateViews(db); err != nil { + t.Fatal(err) + } + base := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + stamp := base.Add(time.Hour).UnixNano() + horizon := stamp + d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return horizon, horizon, stamp }} + for _, span := range []rollupTestSpan{ + {namespace: "ns", traceID: "late", spanID: "child", parentSpanID: "parent", service: "late-child", start: base, ingested: stamp}, + {namespace: "ns", traceID: "existing", spanID: "parent", service: "gateway", start: base, ingested: stamp}, + {namespace: "ns", traceID: "existing", spanID: "child", parentSpanID: "parent", service: "existing-child", start: base, ingested: stamp}, + {namespace: "ns", traceID: "tail", spanID: "parent", service: "gateway", start: base.Add(31 * time.Minute), ingested: stamp}, + {namespace: "ns", traceID: "tail", spanID: "child", parentSpanID: "parent", service: "tail-child", start: base.Add(31 * time.Minute), ingested: stamp}, + } { + insertRollupTestSpan(t, db, span) + } + if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || !pending { + t.Fatalf("initial pass pending=%v err=%v", pending, err) + } + // Publish a parent for an already-completed event bucket while the old + // ingestion window is paused. It belongs to the NEXT discovery window. + horizon++ + insertRollupTestSpan(t, db, rollupTestSpan{namespace: "ns", traceID: "late", spanID: "parent", service: "late-gateway", start: base, ingested: horizon}) + if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || !pending { + t.Fatalf("resumed pass pending=%v err=%v", pending, err) + } + if watermark, cursor := readEdgeRollupState(t, d); watermark != stamp || cursor != 0 { + t.Fatalf("resume changed frozen ingestion end: watermark=%d cursor=%d", watermark, cursor) + } + requireRowCount(t, db, "SELECT count(*) FROM edge_rollup WHERE callee=?", "late-child", 0) + if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || pending { + t.Fatalf("repair pass pending=%v err=%v", pending, err) + } + // Repair replaces the complete bucket: both the new relationship and the + // unrelated edge that was already aggregated must appear exactly once. + for _, child := range []string{"late-child", "existing-child", "tail-child"} { + requireRowCount(t, db, "SELECT count(*) FROM edge_rollup WHERE calls=1 AND callee=?", child, 1) + } + if watermark, cursor := readEdgeRollupState(t, d); watermark != horizon || cursor != 0 { + t.Fatalf("repair did not reach publication horizon: watermark=%d cursor=%d", watermark, cursor) + } +} diff --git a/internal/query/file_snapshot_test.go b/internal/query/file_snapshot_test.go index d96512e4..defc570f 100644 --- a/internal/query/file_snapshot_test.go +++ b/internal/query/file_snapshot_test.go @@ -321,7 +321,7 @@ func TestReadCacheSemanticVersionRebuildsAllContributions(t *testing.T) { if _, err := d.RefreshReadCaches(t.Context()); err != nil { t.Fatal(err) } - for _, stmt := range []string{`UPDATE read_trace_candidates SET min_start=0`, `UPDATE read_cache_version SET version=3`, `CREATE TABLE read_endpoints (value INTEGER)`, `CREATE TABLE read_logs (value INTEGER)`, `CREATE TABLE read_obsolete (value INTEGER)`, `CREATE TABLE endpoint_rollup (value INTEGER)`, `INSERT INTO rollup_state VALUES ('endpoint_rollup:prod',1,now())`} { + for _, stmt := range []string{`UPDATE read_trace_candidates SET min_start=0`, `UPDATE read_cache_version SET version=4`, `INSERT INTO read_edge_buckets VALUES ('ns',now(),10)`, `INSERT INTO rollup_state VALUES ('edge_rollup_v2_substart',1,now()),('edge_rollup_v2_subwindow_end',2,now())`, `CREATE TABLE read_endpoints (value INTEGER)`, `CREATE TABLE read_logs (value INTEGER)`, `CREATE TABLE read_obsolete (value INTEGER)`, `CREATE TABLE endpoint_rollup (value INTEGER)`, `INSERT INTO rollup_state VALUES ('endpoint_rollup:prod',1,now())`} { if _, err := d.DB.Exec(stmt); err != nil { t.Fatal(err) } @@ -329,12 +329,19 @@ func TestReadCacheSemanticVersionRebuildsAllContributions(t *testing.T) { if err := CreateCacheTables(d.DB); err != nil { t.Fatal(err) } - for _, table := range []string{"read_batches", "read_trace_parts", "read_trace_candidates"} { + for _, table := range []string{"read_batches", "read_trace_parts", "read_trace_candidates", "read_edge_buckets"} { var count int if err := d.DB.QueryRow("SELECT count(*) FROM " + table).Scan(&count); err != nil || count != 0 { t.Fatalf("%s survived semantic change: %d %v", table, count, err) } } + if watermark, cursor := readEdgeRollupState(t, d); watermark != 0 || cursor != 0 { + t.Fatalf("obsolete edge cursor survived: watermark=%d cursor=%d", watermark, cursor) + } + var resumeEnd int64 + if err := d.DB.QueryRow(`SELECT coalesce(max(last_ingested_unix_nano),0) FROM rollup_state WHERE cache_key=?`, edgeRollupSubWindowKey).Scan(&resumeEnd); err != nil || resumeEnd != 0 { + t.Fatalf("obsolete edge window survived: end=%d err=%v", resumeEnd, err) + } var obsolete int if err := d.DB.QueryRow(`SELECT count(*) FROM duckdb_tables() WHERE table_name IN ('read_obsolete','endpoint_rollup','read_endpoints','read_logs')`).Scan(&obsolete); err != nil || obsolete != 0 { t.Fatalf("obsolete tables=%d: %v", obsolete, err) @@ -350,7 +357,7 @@ func TestReadCacheSemanticVersionRebuildsAllContributions(t *testing.T) { t.Fatalf("rebuilt bound=%d: %v", minStart, err) } var version int - if err := d.DB.QueryRow(`SELECT version FROM read_cache_version`).Scan(&version); err != nil || version != 4 { + if err := d.DB.QueryRow(`SELECT version FROM read_cache_version`).Scan(&version); err != nil || version != 5 { t.Fatalf("version=%d: %v", version, err) } diff --git a/internal/query/sql_boundary_test.go b/internal/query/sql_boundary_test.go index 56f3ac37..274619e1 100644 --- a/internal/query/sql_boundary_test.go +++ b/internal/query/sql_boundary_test.go @@ -33,6 +33,7 @@ func TestSQLBoundaryRejectsHiddenReadsAndMutations(t *testing.T) { "SELECT * FROM read_traces", "SELECT * FROM read_trace_candidates", "SELECT * FROM read_cache_version", + "SELECT * FROM read_edge_buckets", "SELECT * FROM endpoint_rollup", "SELECT current_setting('allowed_directories')", "SELECT getvariable('operator_secret')", @@ -115,6 +116,7 @@ func TestExecuteSQLRejectsEngineMetadataReads(t *testing.T) { "SELECT * FROM read_traces", "SELECT * FROM read_trace_candidates", "SELECT * FROM read_cache_version", + "SELECT * FROM read_edge_buckets", "SELECT * FROM endpoint_rollup", "SELECT current_setting('allowed_directories')", "SELECT getvariable('operator_secret')", diff --git a/ui/host/e2e/system-overview.spec.ts b/ui/host/e2e/system-overview.spec.ts index 8deedadf..395dc182 100644 --- a/ui/host/e2e/system-overview.spec.ts +++ b/ui/host/e2e/system-overview.spec.ts @@ -1,7 +1,7 @@ -import { expect } from "@playwright/test"; +import { expect, type Response } from "@playwright/test"; import { cardSelector, chooseView, readyCard, test } from "./accessibility-support"; import { chartLabel } from "./setup-support"; -import type { Panel, PanelResult } from "../../panels/types"; +import type { DashboardTime, Panel, PanelResult } from "../../panels/types"; import { frameRows } from "../../panels/rows"; import { statusFor } from "../../panels/thresholds"; @@ -20,12 +20,6 @@ test("default system overview shows all six self-observation scenarios within tw if (!window) throw new Error("Global setup did not provide the seed window"); await page.clock.setFixedTime(new Date(window.to)); await page.addInitScript(theme => localStorage.setItem("mantine-color-scheme-value", theme), test.info().project.use.colorScheme!); - const results = new Map(); - page.on("response", async response => { - if (new URL(response.url()).pathname !== "/api/panels/query" || !response.ok()) return; - const body = await response.json() as { results: PanelResult[] }; - for (const result of body.results) results.set(result.id, result); - }); await page.goto("/"); await expect(page).toHaveURL(/\/dashboards\/[^/?]+/); // The server resolves a relative range with its own clock, which keeps @@ -33,6 +27,16 @@ test("default system overview shows all six self-observation scenarios within tw const pinned = new URL(page.url()); pinned.searchParams.set("from", new Date(new Date(window.to).getTime() - 3_600_000).toISOString()); pinned.searchParams.set("to", new Date(window.to).toISOString()); + // Only collect requests for the pinned document. The first document's + // relative-range responses can lose their bodies during navigation. Attach + // before navigating so the pinned document's initial results are included. + const responses: Response[] = []; + const collectResponse = (response: Response) => { + if (new URL(response.url()).pathname !== "/api/panels/query" || !response.ok()) return; + const time = (response.request().postDataJSON() as { time?: DashboardTime }).time; + if (time?.from === pinned.searchParams.get("from") && time?.to === pinned.searchParams.get("to")) responses.push(response); + }; + page.on("response", collectResponse); await page.goto(pinned.pathname + pinned.search); await expect(page.getByText("System overview", { exact: true }).first()).toBeVisible(); // The harness scrolls each ordinary saved panel into view to trigger its @@ -44,6 +48,12 @@ test("default system overview shows all six self-observation scenarios within tw await response; for (const panel of panels) await readyCard(page, panel); } + page.off("response", collectResponse); + const results = new Map(); + for (const response of responses) { + const body = await response.json() as { results: PanelResult[] }; + for (const result of body.results) results.set(result.id, result); + } const hosts = page.locator(cardSelector("hosts")); await hosts.scrollIntoViewIfNeeded();