From 41978482e7b83ee1fc390e5254bfcc6cb5aceacd Mon Sep 17 00:00:00 2001 From: "courtmanr@gmail.com" Date: Wed, 29 Jul 2026 23:17:58 +0100 Subject: [PATCH] Restore cross-platform verification coverage Use a native absolute path in the SMART override test so Windows verifies the same contract as Unix. Move the metrics concurrency proof into the canonical verification artifact and document that live ingestion remains independent from lifecycle maintenance. --- .../subsystems/performance-and-scalability.md | 9 +- .../hostagent/smartctl_standby_guard_test.go | 6 +- .../issue1601_read_concurrency_test.go | 123 ------------------ pkg/metrics/store_additional_test.go | 114 ++++++++++++++++ 4 files changed, 126 insertions(+), 126 deletions(-) delete mode 100644 pkg/metrics/issue1601_read_concurrency_test.go diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index afd715022..1ee3bf824 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -1229,7 +1229,14 @@ aggregate-only min/max columns stay unset. Rollups must not be hard-coded to a 5-minute disk-write cadence: the default cadence should favor set-based transactions, remain bounded by raw retention so data is aggregated before pruning, and stay overrideable for operators who deliberately trade freshness -for lower write frequency. Rollup transactions must also keep a bounded +for lower write frequency. Startup cleanup, rollup, and retention must run on a +lifecycle-owned maintenance worker separate from the worker draining the +bounded live-ingestion queue. SQLite remains the serialization boundary for +their transactions, but maintenance CPU or read work must not stop buffered +writes or `Flush` barriers from progressing; shutdown must join both workers +before closing their shared database. `pkg/metrics/store_additional_test.go` +must prove ingestion remains live while startup maintenance is blocked. Rollup +transactions must also keep a bounded time-window working set and resume from checkpoints across invocations instead of aggregating every pending series and bucket in one global SQLite GROUP BY that can spike RSS on agent-heavy installs. The runtime must also allow an explicit diff --git a/internal/hostagent/smartctl_standby_guard_test.go b/internal/hostagent/smartctl_standby_guard_test.go index 4c1f7162c..b9e7102db 100644 --- a/internal/hostagent/smartctl_standby_guard_test.go +++ b/internal/hostagent/smartctl_standby_guard_test.go @@ -6,6 +6,7 @@ import ( "fmt" "io/fs" "os/exec" + "path/filepath" "slices" "strings" "testing" @@ -193,7 +194,8 @@ func TestIssue1653DSMDeviceGetsSATRetry(t *testing.T) { } func TestIssue1653SmartctlPathOverride(t *testing.T) { - t.Setenv("PULSE_SMARTCTL_PATH", "/opt/syno/bin/smartctl") + configuredPath := filepath.Join(t.TempDir(), "smartctl") + t.Setenv("PULSE_SMARTCTL_PATH", configuredPath) originalLookPath := execLookPath t.Cleanup(func() { execLookPath = originalLookPath }) execLookPath = func(string) (string, error) { @@ -204,7 +206,7 @@ func TestIssue1653SmartctlPathOverride(t *testing.T) { if err != nil { t.Fatalf("resolveSmartctlPath() error = %v", err) } - if path != "/opt/syno/bin/smartctl" { + if path != configuredPath { t.Fatalf("resolveSmartctlPath() = %q", path) } } diff --git a/pkg/metrics/issue1601_read_concurrency_test.go b/pkg/metrics/issue1601_read_concurrency_test.go deleted file mode 100644 index f50e71ba3..000000000 --- a/pkg/metrics/issue1601_read_concurrency_test.go +++ /dev/null @@ -1,123 +0,0 @@ -package metrics - -import ( - "sync" - "testing" - "time" -) - -// A connection pool of one queued every UI history read behind metric-flush -// commits, freezing charts whenever a commit picked up a WAL checkpoint -// (#1601). Reads must proceed on their own connections while a write -// transaction holds the WAL write lock. -func TestQueriesDoNotQueueBehindWriteTransaction(t *testing.T) { - store, err := NewStore(DefaultConfig(t.TempDir())) - if err != nil { - t.Fatalf("NewStore: %v", err) - } - defer store.Close() - - // Let startup maintenance (checkpoint/vacuum) finish so its locks don't - // entangle with the write transaction opened below. - if err := store.WaitForMaintenance(10 * time.Second); err != nil { - t.Fatalf("WaitForMaintenance: %v", err) - } - - if got := store.db.DB.Stats().MaxOpenConnections; got < 2 { - t.Fatalf("MaxOpenConnections = %d, want at least 2 so reads do not serialize behind writes", got) - } - - // Hold the WAL write lock on one connection. - tx, err := store.db.Begin() - if err != nil { - t.Fatalf("Begin: %v", err) - } - defer func() { _ = tx.Rollback() }() - if _, err := tx.Exec( - `INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) - VALUES ('node', 'n1', 'cpu', 1.0, ?, 'raw')`, - time.Now().Unix(), - ); err != nil { - t.Fatalf("write inside transaction: %v", err) - } - - done := make(chan error, 1) - go func() { - _, err := store.Query("node", "n1", "cpu", time.Now().Add(-time.Hour), time.Now(), 60) - done <- err - }() - - select { - case err := <-done: - if err != nil { - t.Fatalf("Query during open write transaction: %v", err) - } - case <-time.After(5 * time.Second): - t.Fatal("Query blocked behind the open write transaction") - } -} - -// Startup maintenance used to execute on the same worker that drained -// writeCh. A large retention/VACUUM pass could therefore fill the bounded -// write queue and drop live samples. Keep maintenance blocked here and prove -// that ingestion and its Flush barrier remain operational. -func TestStartupMaintenanceDoesNotBlockIngestionWorker(t *testing.T) { - maintenanceEntered := make(chan struct{}) - releaseMaintenance := make(chan struct{}) - var releaseOnce sync.Once - release := func() { - releaseOnce.Do(func() { close(releaseMaintenance) }) - } - t.Cleanup(release) - - previousHook := startupMaintenanceHook - startupMaintenanceHook = func() { - close(maintenanceEntered) - <-releaseMaintenance - } - t.Cleanup(func() { startupMaintenanceHook = previousHook }) - - config := DefaultConfig(t.TempDir()) - config.WriteBufferSize = 1 - config.FlushInterval = time.Hour - - store, err := NewStore(config) - if err != nil { - t.Fatalf("NewStore: %v", err) - } - defer store.Close() - - select { - case <-maintenanceEntered: - case <-time.After(5 * time.Second): - t.Fatal("startup maintenance did not begin") - } - - timestamp := time.Now().UTC().Truncate(time.Second) - store.Write("node", "n1", "cpu", 42, timestamp) - - flushed := make(chan struct{}) - go func() { - store.Flush() - close(flushed) - }() - - select { - case <-flushed: - case <-time.After(5 * time.Second): - t.Fatal("metric ingestion blocked behind startup maintenance") - } - - release() - if err := store.WaitForMaintenance(10 * time.Second); err != nil { - t.Fatalf("WaitForMaintenance: %v", err) - } - - points, err := store.Query("node", "n1", "cpu", timestamp.Add(-time.Second), timestamp.Add(time.Second), 0) - if err != nil { - t.Fatalf("Query: %v", err) - } - if len(points) != 1 || points[0].Value != 42 { - t.Fatalf("persisted points = %+v, want one value 42", points) - } -} diff --git a/pkg/metrics/store_additional_test.go b/pkg/metrics/store_additional_test.go index dfaf52e3f..dc63c9a8b 100644 --- a/pkg/metrics/store_additional_test.go +++ b/pkg/metrics/store_additional_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "strings" + "sync" "syscall" "testing" "time" @@ -524,6 +525,119 @@ func TestStoreWaitForMaintenanceWaitsForQueuedStartupWork(t *testing.T) { } } +// A connection pool of one queued every UI history read behind metric-flush +// commits, freezing charts whenever a commit picked up a WAL checkpoint +// (#1601). Reads must proceed on their own connections while a write +// transaction holds the WAL write lock. +func TestQueriesDoNotQueueBehindWriteTransaction(t *testing.T) { + store, err := NewStore(DefaultConfig(t.TempDir())) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + defer store.Close() + + if err := store.WaitForMaintenance(10 * time.Second); err != nil { + t.Fatalf("WaitForMaintenance: %v", err) + } + + if got := store.db.DB.Stats().MaxOpenConnections; got < 2 { + t.Fatalf("MaxOpenConnections = %d, want at least 2 so reads do not serialize behind writes", got) + } + + tx, err := store.db.Begin() + if err != nil { + t.Fatalf("Begin: %v", err) + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.Exec( + `INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) + VALUES ('node', 'n1', 'cpu', 1.0, ?, 'raw')`, + time.Now().Unix(), + ); err != nil { + t.Fatalf("write inside transaction: %v", err) + } + + done := make(chan error, 1) + go func() { + _, err := store.Query("node", "n1", "cpu", time.Now().Add(-time.Hour), time.Now(), 60) + done <- err + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Query during open write transaction: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Query blocked behind the open write transaction") + } +} + +// Startup maintenance used to execute on the same worker that drained +// writeCh. A large retention/VACUUM pass could therefore fill the bounded +// write queue and drop live samples. Keep maintenance blocked here and prove +// that ingestion and its Flush barrier remain operational. +func TestStartupMaintenanceDoesNotBlockIngestionWorker(t *testing.T) { + maintenanceEntered := make(chan struct{}) + releaseMaintenance := make(chan struct{}) + var releaseOnce sync.Once + release := func() { + releaseOnce.Do(func() { close(releaseMaintenance) }) + } + t.Cleanup(release) + + previousHook := startupMaintenanceHook + startupMaintenanceHook = func() { + close(maintenanceEntered) + <-releaseMaintenance + } + t.Cleanup(func() { startupMaintenanceHook = previousHook }) + + config := DefaultConfig(t.TempDir()) + config.WriteBufferSize = 1 + config.FlushInterval = time.Hour + + store, err := NewStore(config) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + defer store.Close() + + select { + case <-maintenanceEntered: + case <-time.After(5 * time.Second): + t.Fatal("startup maintenance did not begin") + } + + timestamp := time.Now().UTC().Truncate(time.Second) + store.Write("node", "n1", "cpu", 42, timestamp) + + flushed := make(chan struct{}) + go func() { + store.Flush() + close(flushed) + }() + + select { + case <-flushed: + case <-time.After(5 * time.Second): + t.Fatal("metric ingestion blocked behind startup maintenance") + } + + release() + if err := store.WaitForMaintenance(10 * time.Second); err != nil { + t.Fatalf("WaitForMaintenance: %v", err) + } + + points, err := store.Query("node", "n1", "cpu", timestamp.Add(-time.Second), timestamp.Add(time.Second), 0) + if err != nil { + t.Fatalf("Query: %v", err) + } + if len(points) != 1 || points[0].Value != 42 { + t.Fatalf("persisted points = %+v, want one value 42", points) + } +} + func TestStoreMigratesLegacyHostResourceTypeToAgent(t *testing.T) { dir := t.TempDir() cfg := DefaultConfig(dir)