diff --git a/pkg/api/indexstore/indexstore.go b/pkg/api/indexstore/indexstore.go index 6708038e..ce9bd3df 100644 --- a/pkg/api/indexstore/indexstore.go +++ b/pkg/api/indexstore/indexstore.go @@ -13,6 +13,7 @@ import ( "github.com/sirupsen/logrus" "gorm.io/driver/postgres" "gorm.io/gorm" + "gorm.io/gorm/clause" ) // Store provides persistence for the indexed benchmark data. @@ -237,15 +238,34 @@ func (s *store) Stop() error { return sqlDB.Close() } +// runUpsertColumns lists the columns a re-index overwrites on conflict. It is +// every column except the primary key (id), the conflict target +// (discovery_path, run_id), and indexed_at. Listing them explicitly (rather +// than UpdateAll) still overwrites fields reset to their zero value (e.g. +// tests_failed back to 0, a cleared status), which a struct-based Assign/Updates +// would skip — while preserving indexed_at as the original first-index time. +// reindexed_at is updated so each re-index records when it happened. +var runUpsertColumns = []string{ + "timestamp", "timestamp_end", "suite_hash", "status", + "termination_reason", "has_result", "instance_id", "client", + "image", "rollback_strategy", "tests_total", "tests_passed", + "tests_failed", "steps_json", "metadata_json", "reindexed_at", +} + // UpsertRun inserts or updates a run record keyed by discovery_path + run_id. +// +// It uses an INSERT ... ON CONFLICT DO UPDATE so a re-index overwrites the run's +// columns, including fields reset to their zero value. The original indexed_at +// is deliberately preserved (see runUpsertColumns); only reindexed_at tracks the +// latest re-index. func (s *store) UpsertRun(ctx context.Context, run *Run) error { - result := s.db.WithContext(ctx). - Where("discovery_path = ? AND run_id = ?", - run.DiscoveryPath, run.RunID). - Assign(run). - FirstOrCreate(run) - if result.Error != nil { - return fmt.Errorf("upserting run: %w", result.Error) + if err := s.db.WithContext(ctx). + Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "discovery_path"}, {Name: "run_id"}}, + DoUpdates: clause.AssignmentColumns(runUpsertColumns), + }). + Create(run).Error; err != nil { + return fmt.Errorf("upserting run: %w", err) } return nil @@ -630,13 +650,21 @@ type clientRunRow struct { func (s *store) ListTestStatsBySuiteRecent( ctx context.Context, suiteHash string, maxRunsPerClient int, ) ([]TestStat, error) { - // Step 1: lightweight query to get distinct client/run combos. + // Step 1: lightweight query to get one row per client/run combo. We group by + // client and run_id (rather than SELECT DISTINCT over run_start too) so that + // a run whose stats carry inconsistent run_start values still counts as a + // single run. Otherwise it would produce several rows, consume several of the + // per-client slots, and evict other recent runs. var rows []clientRunRow if err := s.readDB.WithContext(ctx). Model(&TestStat{}). - Select("DISTINCT client, run_id, run_start"). + Select("client, run_id, MAX(run_start) AS run_start"). Where("suite_hash = ?", suiteHash). - Order("run_start DESC"). + Group("client, run_id"). + // run_id is a deterministic tie-breaker so that, when more runs than the + // per-client cap share the same run_start, the same runs are kept on + // every call and across SQLite and Postgres. + Order("run_start DESC, run_id DESC"). Find(&rows).Error; err != nil { return nil, fmt.Errorf( "listing recent client runs: %w", err, diff --git a/pkg/api/indexstore/indexstore_test.go b/pkg/api/indexstore/indexstore_test.go index 553f00ee..ed888151 100644 --- a/pkg/api/indexstore/indexstore_test.go +++ b/pkg/api/indexstore/indexstore_test.go @@ -112,10 +112,10 @@ func TestStore_UpsertRunIdempotent(t *testing.T) { require.NoError(t, err) require.Len(t, runs, 1, "upsert must not duplicate the row") - // The original values are preserved (first-write-wins with the - // current Assign+FirstOrCreate implementation). - assert.Equal(t, "running", runs[0].Status) - assert.Equal(t, 5, runs[0].TestsTotal) + // The second upsert overwrites the row (last-write-wins), so a re-index can + // correct an already-stored run. + assert.Equal(t, "completed", runs[0].Status) + assert.Equal(t, 10, runs[0].TestsTotal) } func TestStore_ListRunIDs(t *testing.T) { diff --git a/pkg/api/indexstore/upsert_recent_test.go b/pkg/api/indexstore/upsert_recent_test.go new file mode 100644 index 00000000..5dfe81c5 --- /dev/null +++ b/pkg/api/indexstore/upsert_recent_test.go @@ -0,0 +1,155 @@ +package indexstore_test + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/benchmarkoor/pkg/api/indexstore" +) + +// A re-index of an existing run must overwrite every field, including fields +// reset to their zero value, and must not create a duplicate row. +func TestUpsertRunUpdatesAllFieldsIncludingZeros(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + + require.NoError(t, s.UpsertRun(ctx, &indexstore.Run{ + DiscoveryPath: "dp", RunID: "run-1", + Status: "completed", Client: "geth", + TestsFailed: 5, TestsPassed: 10, TimestampEnd: 999, HasResult: true, + })) + + require.NoError(t, s.UpsertRun(ctx, &indexstore.Run{ + DiscoveryPath: "dp", RunID: "run-1", + Status: "failed", Client: "reth", + TestsFailed: 0, TestsPassed: 7, TimestampEnd: 0, HasResult: false, + })) + + got, err := s.GetRunByRunID(ctx, "run-1") + require.NoError(t, err) + + assert.Equal(t, "failed", got.Status) + assert.Equal(t, "reth", got.Client) + assert.Equal(t, 7, got.TestsPassed) + assert.Equal(t, 0, got.TestsFailed, "zero-valued field should persist") + assert.Equal(t, int64(0), got.TimestampEnd, "zero-valued field should persist") + assert.False(t, got.HasResult, "zero-valued field should persist") + + runs, err := s.ListRuns(ctx, "dp") + require.NoError(t, err) + assert.Len(t, runs, 1, "re-index should update in place, not duplicate") +} + +// A re-index must update the run's mutable fields but preserve the original +// indexed_at (first-index time), recording the re-index time only in +// reindexed_at. Overwriting indexed_at would lose the first-index timestamp. +func TestUpsertRunPreservesIndexedAtOnReindex(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + + firstIndexed := time.Unix(1000, 0).UTC() + require.NoError(t, s.UpsertRun(ctx, &indexstore.Run{ + DiscoveryPath: "dp", RunID: "run-1", + Status: "running", Client: "geth", + IndexedAt: firstIndexed, + })) + + reindexed := time.Unix(2000, 0).UTC() + require.NoError(t, s.UpsertRun(ctx, &indexstore.Run{ + DiscoveryPath: "dp", RunID: "run-1", + Status: "completed", Client: "geth", + IndexedAt: reindexed, ReindexedAt: &reindexed, + })) + + got, err := s.GetRunByRunID(ctx, "run-1") + require.NoError(t, err) + + assert.Equal(t, "completed", got.Status, "mutable fields should update") + assert.Equal(t, firstIndexed.Unix(), got.IndexedAt.Unix(), + "indexed_at must remain the original first-index time") + require.NotNil(t, got.ReindexedAt, "reindexed_at should be recorded") + assert.Equal(t, reindexed.Unix(), got.ReindexedAt.Unix(), + "reindexed_at must record the latest re-index") +} + +func TestUpsertRunInsertsNewRun(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + + require.NoError(t, s.UpsertRun(ctx, &indexstore.Run{ + DiscoveryPath: "dp", RunID: "run-1", + Status: "completed", Client: "geth", HasResult: true, + })) + + got, err := s.GetRunByRunID(ctx, "run-1") + require.NoError(t, err) + assert.Equal(t, "completed", got.Status) + assert.Equal(t, "geth", got.Client) + assert.True(t, got.HasResult) +} + +func stat(suite, runID, testName, client string, runStart int64) *indexstore.TestStat { + return &indexstore.TestStat{ + SuiteHash: suite, + RunID: runID, + TestName: testName, + Client: client, + RunStart: runStart, + } +} + +func distinctRunIDs(stats []indexstore.TestStat) []string { + seen := make(map[string]struct{}) + var out []string + + for _, s := range stats { + if _, ok := seen[s.RunID]; !ok { + seen[s.RunID] = struct{}{} + out = append(out, s.RunID) + } + } + + return out +} + +// A run whose stats carry inconsistent run_start values must count as a single +// run, so it does not evict other recent runs from the per-client window. +func TestListRecentCountsInconsistentRunStartAsOneRun(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + suite := "suite-1" + + require.NoError(t, s.BulkUpsertTestStats(ctx, []*indexstore.TestStat{ + stat(suite, "run-A", "t1", "geth", 300), + stat(suite, "run-A", "t2", "geth", 290), // inconsistent run_start for run-A + stat(suite, "run-B", "t1", "geth", 200), + })) + + got, err := s.ListTestStatsBySuiteRecent(ctx, suite, 2) + require.NoError(t, err) + + assert.ElementsMatch(t, []string{"run-A", "run-B"}, distinctRunIDs(got), + "both recent runs should be returned despite run-A's inconsistent run_start") +} + +func TestListRecentRespectsPerClientCap(t *testing.T) { + s := setupTestStore(t) + ctx := context.Background() + suite := "suite-2" + + require.NoError(t, s.BulkUpsertTestStats(ctx, []*indexstore.TestStat{ + stat(suite, "run-A", "t1", "geth", 300), + stat(suite, "run-B", "t1", "geth", 200), + stat(suite, "run-C", "t1", "geth", 100), + })) + + got, err := s.ListTestStatsBySuiteRecent(ctx, suite, 2) + require.NoError(t, err) + + assert.ElementsMatch(t, []string{"run-A", "run-B"}, distinctRunIDs(got), + "only the 2 most recent runs should be returned") +}