From 9811ed1faab48344389ba5c5a31687df272e887d Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Wed, 23 Sep 2026 00:45:27 +0530 Subject: [PATCH] fix: compare only observed baseline sections Missing or unavailable observations previously became false connection drops, standby losses and new-query or archiving events. Check section provenance while retaining observed zero and positive legacy measurements. Regression coverage includes real SQLite history and input immutability. Verification: affected packages pass with the race detector. No collector SQL, dependency or JSON schema changes. Revert this commit to restore the prior behavior. --- cmd/pgbot/diff_availability_test.go | 107 ++++++++++++ internal/diff/availability_test.go | 256 ++++++++++++++++++++++++++++ internal/diff/diff.go | 57 ++++++- 3 files changed, 412 insertions(+), 8 deletions(-) create mode 100644 cmd/pgbot/diff_availability_test.go create mode 100644 internal/diff/availability_test.go diff --git a/cmd/pgbot/diff_availability_test.go b/cmd/pgbot/diff_availability_test.go new file mode 100644 index 0000000..cc80ebf --- /dev/null +++ b/cmd/pgbot/diff_availability_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "path/filepath" + "testing" + "time" + + "github.com/pgrundev/pgbot/internal/model" + "github.com/pgrundev/pgbot/internal/store" +) + +func TestResolveDiffPreservesMissingObservationsThroughSQLite(t *testing.T) { + tests := []struct { + name string + baseline func() *model.Context + current func() *model.Context + checkExact func(*testing.T, *diffResult) + }{ + { + name: "unavailable baseline then collection recovers", + baseline: func() *model.Context { + return &model.Context{ + Health: &model.Health{Section: model.Section{Exactness: model.ExactnessUnavailable}}, + Queries: &model.Queries{Section: model.Section{Exactness: model.ExactnessUnavailable}, Enabled: true}, + Replication: &model.Replication{Section: model.Section{Exactness: model.ExactnessUnavailable}}, + Archiver: &model.Archiver{Section: model.Section{Exactness: model.ExactnessUnavailable}}, + } + }, + current: func() *model.Context { + return &model.Context{ + Health: &model.Health{Section: model.Section{Exactness: model.ExactnessSampled}, Connections: 100}, + Queries: &model.Queries{Section: model.Section{Exactness: model.ExactnessCumulative}, Enabled: true, Top: []model.QueryStat{ + {QueryID: 1, MeanMS: 10, TotalMS: 100}, + }}, + Replication: &model.Replication{Section: model.Section{Exactness: model.ExactnessScraped}, Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + Archiver: &model.Archiver{Section: model.Section{Exactness: model.ExactnessScraped}, FailedCount: 3}, + } + }, + checkExact: func(t *testing.T, r *diffResult) { + if r.Baseline.Context.Queries.Exactness != model.ExactnessUnavailable || + r.Baseline.Context.Archiver.Exactness != model.ExactnessUnavailable { + t.Fatal("SQLite JSON round trip lost baseline unavailability") + } + }, + }, + { + name: "current collection fails after a measured baseline", + baseline: func() *model.Context { + return &model.Context{ + Health: &model.Health{Section: model.Section{Exactness: model.ExactnessSampled}, Connections: 100}, + Replication: &model.Replication{Section: model.Section{Exactness: model.ExactnessScraped}, Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + } + }, + current: func() *model.Context { + return &model.Context{ + Health: &model.Health{Section: model.Section{Exactness: model.ExactnessUnavailable}}, + Replication: &model.Replication{Section: model.Section{Exactness: model.ExactnessUnavailable}}, + } + }, + checkExact: func(t *testing.T, r *diffResult) { + if r.Current.Context.Health.Exactness != model.ExactnessUnavailable || + r.Current.Context.Replication.Exactness != model.ExactnessUnavailable { + t.Fatal("SQLite JSON round trip lost current unavailability") + } + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "baselines.db") + st, err := store.Open(path) + if err != nil { + t.Fatal(err) + } + baseAt := time.Now().UTC().Add(-2 * time.Hour).Truncate(time.Second) + base := tt.baseline() + base.SchemaVersion = model.SchemaVersion + base.Fingerprint = "availability-test" + base.CollectedAt = baseAt + base.Server.Database = "app" + current := tt.current() + current.SchemaVersion = model.SchemaVersion + current.Fingerprint = base.Fingerprint + current.CollectedAt = baseAt.Add(time.Hour) + current.Server.Database = "app" + if _, err := st.Save(base); err != nil { + t.Fatal(err) + } + if _, err := st.Save(current); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + + r, err := resolveDiff(path, base.Fingerprint, 30*time.Minute) + if err != nil { + t.Fatal(err) + } + if got := deltasOrEmpty(r.Deltas); len(got) != 0 { + t.Fatalf("persisted missing observations must not become changes, got %+v", got) + } + tt.checkExact(t, r) + }) + } +} diff --git a/internal/diff/availability_test.go b/internal/diff/availability_test.go new file mode 100644 index 0000000..69df4b4 --- /dev/null +++ b/internal/diff/availability_test.go @@ -0,0 +1,256 @@ +package diff + +import ( + "bytes" + "encoding/json" + "reflect" + "testing" + "time" + + "github.com/pgrundev/pgbot/internal/model" +) + +func section(exactness string) model.Section { + return model.Section{Exactness: exactness} +} + +func ids(d *model.Deltas) []string { + out := make([]string, 0, len(d.Changes)) + for _, delta := range d.Changes { + out = append(out, delta.ID) + } + return out +} + +func TestComputeMissingSectionsAreNotMeasurements(t *testing.T) { + positive := func() *model.Context { + return &model.Context{ + CollectedAt: time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC), + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 100}, + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true, Top: []model.QueryStat{ + {QueryID: 1, MeanMS: 10, TotalMS: 100}, + }}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + Archiver: &model.Archiver{Section: section(model.ExactnessScraped), FailedCount: 5}, + } + } + unavailable := func() *model.Context { + return &model.Context{ + CollectedAt: time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC), + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessUnavailable)}, + Queries: &model.Queries{Section: section(model.ExactnessUnavailable), Enabled: true}, + Replication: &model.Replication{Section: section(model.ExactnessUnavailable)}, + Archiver: &model.Archiver{Section: section(model.ExactnessUnavailable)}, + } + } + + tests := []struct { + name string + base *model.Context + now *model.Context + }{ + {name: "current unavailable", base: positive(), now: unavailable()}, + {name: "baseline unavailable then collection recovers", base: unavailable(), now: positive()}, + {name: "current sections omitted", base: positive(), now: &model.Context{}}, + {name: "baseline sections omitted", base: &model.Context{}, now: positive()}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + d := Compute(tt.now, &Baseline{CollectedAt: tt.base.CollectedAt, Context: tt.base}, nil) + if len(d.Changes) != 0 { + t.Fatalf("missing observations must not produce changes, got %v", ids(d)) + } + }) + } +} + +func TestComputeSuccessfulEmptySectionsRemainMeasurements(t *testing.T) { + baseAt := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + base := &model.Context{ + CollectedAt: baseAt, + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 100}, + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + Archiver: &model.Archiver{Section: section(model.ExactnessScraped)}, + } + now := &model.Context{ + CollectedAt: baseAt.Add(time.Hour), + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 0}, + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true, Top: []model.QueryStat{ + {QueryID: 2, MeanMS: 10, TotalMS: 200}, + }}, + Replication: &model.Replication{Section: section(model.ExactnessScraped)}, + Archiver: &model.Archiver{Section: section(model.ExactnessScraped), FailedCount: 2}, + } + + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + for _, id := range []string{"health.connections", "query.new", "replication.standby_gone", "archiver.failed_count"} { + if change(d, id) == nil { + t.Errorf("successful empty/positive comparison should produce %s; got %v", id, ids(d)) + } + } +} + +func TestComputeGenuineLossAndAdditionRemainVisible(t *testing.T) { + baseAt := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + base := &model.Context{ + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true, Top: []model.QueryStat{ + {QueryID: 1, MeanMS: 10, TotalMS: 100}, + }}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{ + {AppName: "standby-a"}, {AppName: "standby-b"}, + }}, + } + now := &model.Context{ + CollectedAt: baseAt.Add(time.Hour), + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true, Top: []model.QueryStat{ + {QueryID: 1, MeanMS: 10, TotalMS: 150}, {QueryID: 2, MeanMS: 8, TotalMS: 80}, + }}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{ + {AppName: "standby-b"}, {AppName: "standby-c"}, + }}, + } + + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + if got := change(d, "replication.standby_gone"); got == nil || got.Subject != "standby-a" { + t.Fatalf("expected the genuine standby loss, got %+v", got) + } + if got := change(d, "query.new"); got == nil || got.Subject != shortID(2) { + t.Fatalf("expected the genuine top-query addition, got %+v", got) + } + if len(d.Changes) != 2 { + t.Fatalf("standby/query additions must not hide or invent changes, got %v", ids(d)) + } +} + +func TestComputeUnavailableHealthCanRetainMeasuredConnections(t *testing.T) { + baseAt := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + base := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 100}, + } + now := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessUnavailable), Connections: 20}, + } + + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + if got := change(d, "health.connections"); got == nil || got.Before != 100 || got.After != 20 { + t.Fatalf("a retained positive connection gauge remains comparable despite rate unavailability, got %+v", got) + } +} + +func TestComputeLegacySectionsNeedEvidenceBeforeAbsenceIsTrusted(t *testing.T) { + baseAt := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + + t.Run("empty legacy baseline does not make current data new", func(t *testing.T) { + base := &model.Context{ + Health: &model.Health{}, Queries: &model.Queries{Enabled: true}, + Replication: &model.Replication{}, Archiver: &model.Archiver{}, + } + now := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 100}, + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true, Top: []model.QueryStat{ + {QueryID: 1, MeanMS: 10, TotalMS: 100}, + }}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + Archiver: &model.Archiver{Section: section(model.ExactnessScraped), FailedCount: 2}, + } + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + if len(d.Changes) != 0 { + t.Fatalf("an empty legacy section is unknown, not known-empty; got %v", ids(d)) + } + }) + + t.Run("empty legacy current does not imply loss", func(t *testing.T) { + base := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 100}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + } + now := &model.Context{Server: model.ServerInfo{Database: "app"}, Health: &model.Health{}, Replication: &model.Replication{}} + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + if len(d.Changes) != 0 { + t.Fatalf("an empty legacy current section is unknown, not known-empty; got %v", ids(d)) + } + }) + + t.Run("positive legacy values remain usable", func(t *testing.T) { + base := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Connections: 100}, + Queries: &model.Queries{Enabled: true, Top: []model.QueryStat{{QueryID: 1, MeanMS: 10, TotalMS: 100}}}, + Replication: &model.Replication{Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + Archiver: &model.Archiver{FailedCount: 5}, + } + now := &model.Context{ + CollectedAt: baseAt.Add(time.Hour), Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 20}, + Queries: &model.Queries{Section: section(model.ExactnessCumulative), Enabled: true, Top: []model.QueryStat{ + {QueryID: 1, MeanMS: 10, TotalMS: 150}, {QueryID: 2, MeanMS: 8, TotalMS: 80}, + }}, + Replication: &model.Replication{Section: section(model.ExactnessScraped)}, + Archiver: &model.Archiver{Section: section(model.ExactnessScraped), FailedCount: 6}, + } + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + for _, id := range []string{"health.connections", "query.new", "replication.standby_gone", "archiver.failed_count"} { + if change(d, id) == nil { + t.Errorf("positive legacy evidence should preserve %s comparisons; got %v", id, ids(d)) + } + } + }) + + t.Run("legacy health rates prove a zero gauge was collected", func(t *testing.T) { + zero := 0.0 + base := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Connections: 100}, + } + now := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Connections: 0, TPS: &zero}, + } + d := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + if change(d, "health.connections") == nil { + t.Fatalf("a legacy rate is positive evidence that the zero connection gauge was collected; got %v", ids(d)) + } + }) +} + +func TestComputeAvailabilityIsDeterministicAndDoesNotMutateContexts(t *testing.T) { + baseAt := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + base := &model.Context{ + Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessSampled), Connections: 100}, + Replication: &model.Replication{Section: section(model.ExactnessScraped), Replicas: []model.ReplicaRow{{AppName: "standby-a"}}}, + } + now := &model.Context{ + CollectedAt: baseAt.Add(time.Hour), Server: model.ServerInfo{Database: "app"}, + Health: &model.Health{Section: section(model.ExactnessUnavailable)}, + Replication: &model.Replication{Section: section(model.ExactnessUnavailable)}, + } + baseBefore, err := json.Marshal(base) + if err != nil { + t.Fatal(err) + } + nowBefore, err := json.Marshal(now) + if err != nil { + t.Fatal(err) + } + + first := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + second := Compute(now, &Baseline{CollectedAt: baseAt, Context: base}, nil) + if !reflect.DeepEqual(first, second) { + t.Fatalf("same inputs produced different deltas:\nfirst: %+v\nsecond: %+v", first, second) + } + baseAfter, _ := json.Marshal(base) + nowAfter, _ := json.Marshal(now) + if !bytes.Equal(baseBefore, baseAfter) || !bytes.Equal(nowBefore, nowAfter) { + t.Fatal("Compute mutated one of its Context inputs") + } +} diff --git a/internal/diff/diff.go b/internal/diff/diff.go index 8b47002..0325d5d 100644 --- a/internal/diff/diff.go +++ b/internal/diff/diff.go @@ -55,7 +55,7 @@ func Compute(now *model.Context, primary *Baseline, _ *Baseline) *model.Deltas { // replicaGoneDelta flags a standby that was streaming at the baseline but is // absent from pg_stat_replication now — a silent disconnect only history sees. func replicaGoneDelta(now, base *model.Context, baseAt time.Time, d *model.Deltas) { - if now.Replication == nil || base.Replication == nil { + if !replicationObserved(now.Replication) || !replicationObserved(base.Replication) { return } present := map[string]bool{} @@ -86,7 +86,7 @@ func standbyKey(r model.ReplicaRow) string { // point-in-time last_failed_time < last_archived_time comparison misses. Only a // genuine increase (not a counter reset going backwards) is reported. func archiverFailedDelta(now, base *model.Context, baseAt time.Time, d *model.Deltas) { - if now.Archiver == nil || base.Archiver == nil { + if !archiverObserved(now.Archiver) || !archiverObserved(base.Archiver) { return } if inc := now.Archiver.FailedCount - base.Archiver.FailedCount; inc > 0 { @@ -101,14 +101,12 @@ func archiverFailedDelta(now, base *model.Context, baseAt time.Time, d *model.De } func queryDeltas(now, base *model.Context, baseAt time.Time, d *model.Deltas) { - if now.Queries == nil || !now.Queries.Enabled { + if !queriesObserved(now.Queries) || !queriesObserved(base.Queries) { return } baseByID := map[int64]model.QueryStat{} - if base.Queries != nil { - for _, q := range base.Queries.Top { - baseByID[q.QueryID] = q - } + for _, q := range base.Queries.Top { + baseByID[q.QueryID] = q } for _, q := range now.Queries.Top { prev, seen := baseByID[q.QueryID] @@ -187,7 +185,7 @@ func dbSizeDelta(now, base *model.Context, baseAt time.Time, d *model.Deltas) { } func connectionDelta(now, base *model.Context, baseAt time.Time, d *model.Deltas) { - if now.Health == nil || base.Health == nil { + if !connectionsObserved(now.Health) || !connectionsObserved(base.Health) { return } before, after := base.Health.Connections, now.Health.Connections @@ -207,6 +205,49 @@ func connectionDelta(now, base *model.Context, baseAt time.Time, d *model.Deltas } } +// Older stored contexts can lack exactness. Preserve their positive evidence, +// but never turn an all-default legacy section into a confidently empty sample. +func replicationObserved(r *model.Replication) bool { + if r == nil || r.Exactness == model.ExactnessUnavailable { + return false + } + return r.Exactness != "" || r.IsReplica || r.ReceiverLagSec != nil || + len(r.Replicas) > 0 || len(r.Slots) > 0 || len(r.Subscriptions) > 0 +} + +func archiverObserved(a *model.Archiver) bool { + if a == nil || a.Exactness == model.ExactnessUnavailable { + return false + } + return a.Exactness != "" || a.ArchivedCount != 0 || a.LastArchivedWAL != "" || + a.LastArchivedTime != nil || a.FailedCount != 0 || a.LastFailedWAL != "" || + a.LastFailedTime != nil || a.StatsReset != nil || a.HasArchiveCommand +} + +func queriesObserved(q *model.Queries) bool { + if q == nil || !q.Enabled || q.Exactness == model.ExactnessUnavailable { + return false + } + return q.Exactness != "" || q.TotalExecMS != 0 || q.PgssDealloc != 0 || + q.PgssCount != 0 || q.PgssMax != 0 || len(q.Top) > 0 +} + +func connectionsObserved(h *model.Health) bool { + if h == nil { + return false + } + if h.Connections > 0 { + return true + } + if h.Exactness == model.ExactnessUnavailable { + return false + } + return h.Exactness != "" || h.TPS != nil || h.CommitsPerSec != nil || + h.RollbacksPerSec != nil || h.RollbackRatio != nil || h.CacheHitRatio != nil || + h.CacheBlocks != nil || h.DeadlocksPerMin != nil || h.TempBytesPerSec != nil || + h.TupReturnedPerS != nil || h.TupWrittenPerS != nil +} + func deadRatioDelta(now, base *model.Context, baseAt time.Time, d *model.Deltas) { if now.Tables == nil || base.Tables == nil { return