From 249ab8dbd28f541380e167325b8f428102af27d8 Mon Sep 17 00:00:00 2001 From: Stefan Date: Tue, 11 Aug 2026 09:55:17 +0200 Subject: [PATCH] fix(suite): merge into the stored summary, not just the local one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CreateSuiteOutput enriches an existing summary.json rather than replacing it — it preserves the prior tests[], backfills payload sizes and tx counts from the step files, and folds in opcode data. But it only ever looked for that prior summary on local disk. Every CI worker starts a job with an empty RUNNER_TEMP, so the local suite directory never survives: suiteExists is always false, the merge never runs, and the summary is rebuilt from that one run's inputs and uploaded over whatever the store had accumulated. Two runs of the same suite with different config richness — one with tests.opcode_source set, one without — silently resolve to whichever finished last. Pass the stored summary in as a baseline. The executor fetches it through an optional RemoteSuiteSummary hook, wired to the S3 reader when results upload is configured, so the executor keeps no storage dependency and a local-only run behaves exactly as before. Materialising the step files stays keyed on the local summary alone. A stored summary says nothing about what is on local disk, and conflating the two would leave a wiped worker with nothing to upload — which is how a partially uploaded suite would become permanent. A fetch failure logs and falls back to the previous behaviour: the merge is an enrichment, never a precondition for writing the suite. --- cmd/benchmarkoor/run.go | 30 +++++++++++ pkg/executor/executor.go | 37 ++++++++++++-- pkg/executor/suite.go | 90 +++++++++++++++++++++----------- pkg/executor/suite_test.go | 102 ++++++++++++++++++++++++++++++++++--- 4 files changed, 218 insertions(+), 41 deletions(-) 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) + }) + } +}