diff --git a/cmd/benchmarkoor/run.go b/cmd/benchmarkoor/run.go index 76b062c..13a61d9 100644 --- a/cmd/benchmarkoor/run.go +++ b/cmd/benchmarkoor/run.go @@ -256,6 +256,7 @@ func runBenchmark(cmd *cobra.Command, args []string) error { ResultsOwner: resultsOwner, SystemResourceCollectionEnabled: *cfg.Runner.Benchmark.SystemResourceCollectionEnabled, GitHubToken: cfg.Runner.GitHubToken, + RemoteSuiteSummary: remoteSuiteSummaryFetcher(log, cfg), } exec = executor.NewExecutor(log, execCfg) @@ -444,6 +445,35 @@ func generateResultsIndex( } } +// remoteSuiteSummaryFetcher returns a reader for the summary.json already in +// the bucket for a suite hash, or nil when S3 upload is not configured. A CI +// worker starts every job with an empty results directory, so without this the +// suite summary is rebuilt from that run alone and overwrites the stored one. +func remoteSuiteSummaryFetcher( + log logrus.FieldLogger, + cfg *config.Config, +) func(context.Context, string) ([]byte, error) { + if cfg.Runner.Benchmark.ResultsUpload == nil || + cfg.Runner.Benchmark.ResultsUpload.S3 == nil || + !cfg.Runner.Benchmark.ResultsUpload.S3.Enabled { + return nil + } + + s3Cfg := cfg.Runner.Benchmark.ResultsUpload.S3 + + prefix := s3Cfg.Prefix + if prefix == "" { + prefix = "results" + } + + reader := upload.NewS3Reader(log, s3Cfg) + suitesBase := strings.TrimRight(prefix, "/") + "/suites/" + + return func(ctx context.Context, hash string) ([]byte, error) { + return reader.GetObject(ctx, suitesBase+hash+"/summary.json") + } +} + // generateResultsIndexLocal generates index.json from a local results directory. func generateResultsIndexLocal( cfg *config.Config, diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index d19983e..130505c 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -143,6 +143,12 @@ type Config struct { ResultsOwner *fsutil.OwnerConfig // Optional file ownership for results directory SystemResourceCollectionEnabled bool // Enable system resource collection (cgroups/Docker Stats) GitHubToken string // Optional GitHub token for API-based artifact downloads + // RemoteSuiteSummary returns the summary.json remote storage already holds + // for a suite hash, or nil when there is none. A worker whose results + // directory is wiped between jobs — every CI runner — has no local suite to + // merge into, so without this it regenerates the summary from scratch and + // overwrites whatever the store had learned from richer runs. + RemoteSuiteSummary func(ctx context.Context, hash string) ([]byte, error) } // NewExecutor creates a new executor instance. @@ -227,7 +233,7 @@ func (e *executor) Start(ctx context.Context) error { // Create suite output if results directory is configured. if e.cfg.ResultsDir != "" { - if err := e.createSuiteOutput(); err != nil { + if err := e.createSuiteOutput(ctx); err != nil { return fmt.Errorf("creating suite output: %w", err) } } @@ -236,7 +242,7 @@ func (e *executor) Start(ctx context.Context) error { } // createSuiteOutput computes hash and creates suite directory. -func (e *executor) createSuiteOutput() error { +func (e *executor) createSuiteOutput(ctx context.Context) error { // Compute suite hash from file contents. hash, err := ComputeSuiteHash(e.prepared) if err != nil { @@ -259,8 +265,12 @@ func (e *executor) createSuiteOutput() error { Metadata: e.cfg.Metadata, } - // Create suite output directory. - if err := CreateSuiteOutput(e.log, e.cfg.ResultsDir, hash, suiteInfo, e.prepared, e.cfg.ResultsOwner); err != nil { + // Create suite output directory, merging into what the store already knows + // for this hash so a run on a wiped worker enriches rather than replaces. + if err := CreateSuiteOutput( + e.log, e.cfg.ResultsDir, hash, suiteInfo, e.prepared, e.cfg.ResultsOwner, + e.remoteSuiteSummary(ctx, hash), + ); err != nil { return fmt.Errorf("creating suite output: %w", err) } @@ -273,6 +283,25 @@ func (e *executor) createSuiteOutput() error { return nil } +// remoteSuiteSummary fetches the stored summary for a hash. A failure is not +// fatal: it costs the merge, and the run still writes a summary built from what +// it knows, exactly as it did before this existed. +func (e *executor) remoteSuiteSummary(ctx context.Context, hash string) []byte { + if e.cfg.RemoteSuiteSummary == nil { + return nil + } + + data, err := e.cfg.RemoteSuiteSummary(ctx, hash) + if err != nil { + e.log.WithError(err).WithField("hash", hash). + Warn("Failed to read stored suite summary; building it from this run alone") + + return nil + } + + return data +} + // Stop cleans up the executor. func (e *executor) Stop() error { if e.source != nil { diff --git a/pkg/executor/suite.go b/pkg/executor/suite.go index 9b6a10a..e00f9b3 100644 --- a/pkg/executor/suite.go +++ b/pkg/executor/suite.go @@ -101,6 +101,19 @@ type SuiteTest struct { TxCounts *TxCounts `json:"tx_counts,omitempty"` } +// validSummary reports whether data is a summary.json describing a built +// suite. A bare or truncated one carries no tests and must not be merged into +// — that is what leaves "tests": null behind. +func validSummary(data []byte) bool { + if len(data) == 0 { + return false + } + + var s SuiteInfo + + return json.Unmarshal(data, &s) == nil && len(s.Tests) > 0 +} + // ComputeSuiteHash computes a hash of all test file contents. func ComputeSuiteHash(prepared *PreparedSource) (string, error) { h := sha256.New() @@ -159,25 +172,40 @@ func getStepContent(step *StepFile) ([]byte, error) { } // CreateSuiteOutput creates the suite directory structure with copied files and summary. +// remoteSummary is the summary.json already held by remote storage for this +// hash, or nil. It seeds the merge on workers whose results directory does not +// survive between jobs; the files still get materialised either way. func CreateSuiteOutput( log logrus.FieldLogger, resultsDir, hash string, info *SuiteInfo, prepared *PreparedSource, owner *fsutil.OwnerConfig, + remoteSummary []byte, ) error { suiteDir := filepath.Join(resultsDir, "suites", hash) - suiteExists := false - // Treat the suite as already built only when a complete summary.json with // tests is present. A bare or partial directory — e.g. left behind by a run // that aborted mid-creation — must be rebuilt; otherwise info.Tests stays // nil and summary.json gets (re)written with "tests": null. - if data, err := os.ReadFile(filepath.Join(suiteDir, "summary.json")); err == nil { - var existing SuiteInfo - if json.Unmarshal(data, &existing) == nil && len(existing.Tests) > 0 { - suiteExists = true + localSummary, _ := os.ReadFile(filepath.Join(suiteDir, "summary.json")) + suiteExists := validSummary(localSummary) + + // What to merge into: this worker's own copy when it has one, otherwise + // whatever the store holds. Whether the files need materialising is a + // separate question, decided by suiteExists alone — a stored summary says + // nothing about what is on local disk. + baseline := localSummary + + if !suiteExists { + baseline = nil + + if validSummary(remoteSummary) { + baseline = remoteSummary + + log.WithField("bytes", len(remoteSummary)). + Debug("Merging into the stored suite summary") } } @@ -335,37 +363,37 @@ func CreateSuiteOutput( // runs without affecting the suite hash, so we update it every time. summaryPath := filepath.Join(suiteDir, "summary.json") - // If the suite already existed, read the existing summary to preserve - // test/step file references, then overlay the new info fields. - if suiteExists { - existingData, readErr := os.ReadFile(summaryPath) - if readErr == nil { - var existing SuiteInfo - if jsonErr := json.Unmarshal(existingData, &existing); jsonErr == nil { + // Overlay the new info onto the prior summary — the local one when this + // worker already built the suite, otherwise the stored one — so fields an + // earlier run derived from richer inputs survive. + if baseline != nil { + var existing SuiteInfo + if jsonErr := json.Unmarshal(baseline, &existing); jsonErr == nil { + if len(existing.PreRunSteps) > 0 { info.PreRunSteps = existing.PreRunSteps + } - // Merge opcode data from prepared tests into existing entries. - mergeOpcodeData(existing.Tests, prepared) - - lineProvider := func(testName string, step StepKind) []string { - reqPath := filepath.Join(suiteDir, sanitizeResultPath(testName), string(step)+".request") - data, err := os.ReadFile(reqPath) - if err != nil { - // Missing files are normal — most tests don't have setup/cleanup. - // Only warn for the test step where absence is genuinely unexpected. - if step == StepKindTest && !os.IsNotExist(err) { - log.WithError(err).WithField("path", reqPath).Warn("Failed to read test.request for payload-size merge") - } - return nil + // Merge opcode data from prepared tests into existing entries. + mergeOpcodeData(existing.Tests, prepared) + + lineProvider := func(testName string, step StepKind) []string { + reqPath := filepath.Join(suiteDir, sanitizeResultPath(testName), string(step)+".request") + data, err := os.ReadFile(reqPath) + if err != nil { + // Missing files are normal — most tests don't have setup/cleanup. + // Only warn for the test step where absence is genuinely unexpected. + if step == StepKindTest && !os.IsNotExist(err) { + log.WithError(err).WithField("path", reqPath).Warn("Failed to read test.request for payload-size merge") } - return splitNonEmptyLines(string(data)) + return nil } + return splitNonEmptyLines(string(data)) + } - MergePayloadSizes(log, existing.Tests, lineProvider) - MergeTxCounts(log, existing.Tests, lineProvider) + MergePayloadSizes(log, existing.Tests, lineProvider) + MergeTxCounts(log, existing.Tests, lineProvider) - info.Tests = existing.Tests - } + info.Tests = existing.Tests } } diff --git a/pkg/executor/suite_test.go b/pkg/executor/suite_test.go index 7e2e45f..f3b923f 100644 --- a/pkg/executor/suite_test.go +++ b/pkg/executor/suite_test.go @@ -44,7 +44,7 @@ func TestCreateSuiteOutput_WritesPayloadSizes(t *testing.T) { Hash: "deadbeef", } log := logrus.New() - err := CreateSuiteOutput(log, tmp, "deadbeef", info, prepared, nil) + err := CreateSuiteOutput(log, tmp, "deadbeef", info, prepared, nil, nil) require.NoError(t, err) summaryPath := filepath.Join(tmp, "suites", "deadbeef", "summary.json") @@ -84,7 +84,7 @@ func TestCreateSuiteOutput_AggregatesMetadataOpcodeCounts(t *testing.T) { }, } info := &SuiteInfo{Hash: "cafe"} - err := CreateSuiteOutput(logrus.New(), tmp, "cafe", info, prepared, nil) + err := CreateSuiteOutput(logrus.New(), tmp, "cafe", info, prepared, nil, nil) require.NoError(t, err) data, err := os.ReadFile(filepath.Join(tmp, "suites", "cafe", "summary.json")) @@ -167,7 +167,7 @@ func TestCreateSuiteOutput_CopiesEESTMeta(t *testing.T) { } info := &SuiteInfo{Hash: "abc123"} - require.NoError(t, CreateSuiteOutput(logrus.New(), tmp, "abc123", info, prepared, nil)) + require.NoError(t, CreateSuiteOutput(logrus.New(), tmp, "abc123", info, prepared, nil, nil)) suiteMeta := filepath.Join(tmp, "suites", "abc123", ".eest-meta") @@ -203,7 +203,7 @@ func TestCreateSuiteOutput_NoEESTMetaWhenAbsent(t *testing.T) { } info := &SuiteInfo{Hash: "nometa01"} - require.NoError(t, CreateSuiteOutput(logrus.New(), tmp, "nometa01", info, prepared, nil)) + require.NoError(t, CreateSuiteOutput(logrus.New(), tmp, "nometa01", info, prepared, nil, nil)) _, err := os.Stat(filepath.Join(tmp, "suites", "nometa01", ".eest-meta")) assert.True(t, os.IsNotExist(err)) @@ -234,7 +234,7 @@ func TestCreateSuiteOutput_MergesPayloadSizesOnSecondRun(t *testing.T) { // First run — creates the suite and writes initial sizes. log := logrus.New() info1 := &SuiteInfo{Hash: "cafef00d"} - require.NoError(t, CreateSuiteOutput(log, tmp, "cafef00d", info1, prepared, nil)) + require.NoError(t, CreateSuiteOutput(log, tmp, "cafef00d", info1, prepared, nil, nil)) // Simulate a legacy summary: rewrite the file with payload_sizes cleared. summaryPath := filepath.Join(tmp, "suites", "cafef00d", "summary.json") @@ -251,7 +251,7 @@ func TestCreateSuiteOutput_MergesPayloadSizesOnSecondRun(t *testing.T) { // Second run — should detect suite exists, read on-disk test.request, and merge. info2 := &SuiteInfo{Hash: "cafef00d"} - require.NoError(t, CreateSuiteOutput(log, tmp, "cafef00d", info2, prepared, nil)) + require.NoError(t, CreateSuiteOutput(log, tmp, "cafef00d", info2, prepared, nil, nil)) final, err := os.ReadFile(summaryPath) require.NoError(t, err) @@ -263,3 +263,93 @@ func TestCreateSuiteOutput_MergesPayloadSizesOnSecondRun(t *testing.T) { require.Len(t, parsed.Tests[0].PayloadSizes.Test.SSZFull, 1) assert.Greater(t, parsed.Tests[0].PayloadSizes.Test.SSZFull[0], uint64(100), "merge path should backfill sizes") } + +// A CI worker starts each job with an empty results directory, so it has no +// local suite to merge into. Without the stored summary it would rebuild the +// description from its own inputs and overwrite what earlier, richer runs +// contributed — opcode counts from an external source being the clearest case, +// since a run without that config cannot recompute them. +func TestCreateSuiteOutput_MergesStoredSummaryOnWipedWorker(t *testing.T) { + log := logrus.New() + prepared := &PreparedSource{ + Tests: []*TestWithSteps{ + { + Name: "test_opcode_merge", + Test: &StepFile{ + Name: "test_opcode_merge", + Provider: &inlineProvider{lines: []string{minimalDenebRequest(t)}}, + }, + }, + }, + } + + // A previous run, on some other worker, recorded opcode counts. + stored, err := json.Marshal(&SuiteInfo{ + Hash: "beefcafe", + Tests: []SuiteTest{{ + Name: "test_opcode_merge", + OpcodeCount: map[string]int{"PUSH1": 42}, + }}, + }) + require.NoError(t, err) + + // This run gets a fresh results dir and no opcode source of its own. + tmp := t.TempDir() + info := &SuiteInfo{Hash: "beefcafe"} + require.NoError(t, CreateSuiteOutput(log, tmp, "beefcafe", info, prepared, nil, stored)) + + summaryPath := filepath.Join(tmp, "suites", "beefcafe", "summary.json") + data, err := os.ReadFile(summaryPath) + require.NoError(t, err) + + var parsed SuiteInfo + require.NoError(t, json.Unmarshal(data, &parsed)) + require.Len(t, parsed.Tests, 1) + assert.Equal(t, map[string]int{"PUSH1": 42}, parsed.Tests[0].OpcodeCount, + "stored opcode counts must survive a run that cannot recompute them") + + // The step files are still materialised: a stored summary says nothing + // about what is on local disk, and the upload has to have bytes to send. + assert.FileExists(t, filepath.Join(tmp, "suites", "beefcafe", "test_opcode_merge", "test.request")) + + // And the merge still backfills from those materialised files. + require.NotNil(t, parsed.Tests[0].PayloadSizes) + assert.NotNil(t, parsed.Tests[0].PayloadSizes.Test) +} + +// A truncated or test-less stored summary must not be merged into — that is +// what leaves "tests": null behind. +func TestCreateSuiteOutput_IgnoresUnusableStoredSummary(t *testing.T) { + log := logrus.New() + prepared := &PreparedSource{ + Tests: []*TestWithSteps{ + { + Name: "test_ignore_bad", + Test: &StepFile{ + Name: "test_ignore_bad", + Provider: &inlineProvider{lines: []string{minimalDenebRequest(t)}}, + }, + }, + }, + } + + for name, stored := range map[string][]byte{ + "truncated": []byte(`{"hash":"d00d","tests":`), + "no tests": []byte(`{"hash":"d00d"}`), + "empty": nil, + } { + t.Run(name, func(t *testing.T) { + tmp := t.TempDir() + info := &SuiteInfo{Hash: "d00d"} + require.NoError(t, CreateSuiteOutput(log, tmp, "d00d", info, prepared, nil, stored)) + + data, err := os.ReadFile(filepath.Join(tmp, "suites", "d00d", "summary.json")) + require.NoError(t, err) + + var parsed SuiteInfo + require.NoError(t, json.Unmarshal(data, &parsed)) + require.Len(t, parsed.Tests, 1, "falls back to this run's own description") + assert.Equal(t, "test_ignore_bad", parsed.Tests[0].Name) + }) + } +}