fix(indexstore): make UpsertRun overwrite existing rows and stop dropping recent runs

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,

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) {

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")
+}