From 1b409899e8b053b2405806c005f65eaa9cf3f739 Mon Sep 17 00:00:00 2001 From: Marwen Abid Date: Sat, 1 Aug 2026 14:57:12 -0700 Subject: [PATCH 1/2] =?UTF-8?q?runner:=20run=20package=20=E2=80=94=20execu?= =?UTF-8?q?tor,=20leg.json=20sentinels,=20and=20resume?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sequential executor over the plan: leg.json completion sentinels in every timed --out dir, --resume that trusts them, keep-going semantics, dataset preparation, and the persistent stellar-rpc source/build clone. --- runner/internal/run/dataset.go | 177 +++++++ runner/internal/run/dataset_test.go | 248 ++++++++++ runner/internal/run/doc.go | 3 + runner/internal/run/lock.go | 39 ++ runner/internal/run/log.go | 26 + runner/internal/run/resume.go | 140 ++++++ runner/internal/run/run.go | 376 +++++++++++++++ runner/internal/run/run_test.go | 707 ++++++++++++++++++++++++++++ runner/internal/run/source.go | 57 +++ runner/internal/run/source_test.go | 211 +++++++++ 10 files changed, 1984 insertions(+) create mode 100644 runner/internal/run/dataset.go create mode 100644 runner/internal/run/dataset_test.go create mode 100644 runner/internal/run/doc.go create mode 100644 runner/internal/run/lock.go create mode 100644 runner/internal/run/log.go create mode 100644 runner/internal/run/resume.go create mode 100644 runner/internal/run/run.go create mode 100644 runner/internal/run/run_test.go create mode 100644 runner/internal/run/source.go create mode 100644 runner/internal/run/source_test.go diff --git a/runner/internal/run/dataset.go b/runner/internal/run/dataset.go new file mode 100644 index 0000000..396b7dd --- /dev/null +++ b/runner/internal/run/dataset.go @@ -0,0 +1,177 @@ +package run + +import ( + "errors" + "fmt" + "io" + "os" + "path/filepath" + + "github.com/stellar/stellar-rpc-benchmarks/runner/internal/config" + "github.com/stellar/stellar-rpc-benchmarks/runner/internal/plan" +) + +// runDataset converges one dataset on a local cold pack root, whatever kind it +// is: the legs downstream all read /ledgers and neither know nor care +// whether it was fetched, backfilled, generated, or already there. +// +// Everything a kind materializes lands in .partial and is renamed onto +// only once whole, so an interrupted preparation can never be mistaken +// for a finished one — the golden-present check below is exactly that +// distinction, and `rm -rf ` is the documented lever to force a re-fetch. +func runDataset(s plan.Step, opts Options) StepResult { + if s.Dataset == nil { + return failure(s, errors.New("dataset step has no dataset spec")) + } + if err := prepareDataset(s, opts.Output); err != nil { + return failure(s, err) + } + return StepResult{ID: s.ID, Status: StatusOK} +} + +func prepareDataset(s plan.Step, out io.Writer) error { + d := s.Dataset + partial := d.Root + ".partial" + + switch d.Kind { + case config.KindPacksLocal: + Notef(out, "dataset %s: local cold pack root %s", d.Name, d.Root) + if !dirExists(filepath.Join(d.Root, "ledgers")) { + return fmt.Errorf("dataset '%s': %s/ledgers not found — location must be a cold pack root", d.Name, d.Root) + } + + case config.KindPacksGS: + if goldenPresent(d.Root) { + Notef(out, "dataset %s: golden packs already at %s — skipping fetch", d.Name, d.Root) + break + } + Notef(out, "dataset %s: fetch %s", d.Name, d.Location) + // golden_present was false, so root is absent or an empty leftover, and + // the plan's pre_clean says to clear it — but not the partial, which + // rsync resumes into. + if err := preClean(s, out); err != nil { + return err + } + if err := mkdirAll(out, partial); err != nil { + return err + } + if err := runDatasetCommands(s, out); err != nil { + return err + } + if err := rename(out, partial, d.Root); err != nil { + return err + } + + case config.KindBSBS3: + if goldenPresent(d.Root) { + Notef(out, "dataset %s: golden packs already at %s — skipping backfill", d.Name, d.Root) + break + } + Notef(out, "dataset %s: golden backfill of %s from S3 (untimed)", d.Name, d.Location) + if err := preClean(s, out); err != nil { + return err + } + // One untimed cold backfill per chunk. The step's env carries + // AWS_EC2_METADATA_DISABLED=true: without it the SDK signs requests + // with the machine's IAM role and the public bucket 403s, but setting + // it for the whole campaign would also hide those same instance-role + // credentials from the publish step's `aws s3` calls. + if err := runDatasetCommands(s, out); err != nil { + return err + } + if err := rename(out, partial, d.Root); err != nil { + return err + } + + case config.KindFixture: + if goldenPresent(d.Root) { + Notef(out, "dataset %s: golden packs already at %s — skipping generation", d.Name, d.Root) + break + } + if d.Stage == "" { + // The generation commands write into the staging tree, and the + // plan's pre_clean is derived from it; a spec without one describes + // a preparation nobody can reason about. + return fmt.Errorf("dataset '%s': fixture step has no staging pack dir", d.Name) + } + Notef(out, "dataset %s: generate a fixture pack tree", d.Name) + if err := preClean(s, out); err != nil { + return err + } + // Generate every chunk into the staging pack tree, then freeze every + // chunk into the golden packs — both untimed. + if err := runDatasetCommands(s, out); err != nil { + return err + } + if err := rename(out, partial, d.Root); err != nil { + return err + } + + default: + return fmt.Errorf("dataset '%s': unknown kind %q", d.Name, d.Kind) + } + + if !dirExists(filepath.Join(d.Root, "ledgers")) { + return fmt.Errorf("dataset '%s': %s/ledgers missing after preparation", d.Name, d.Root) + } + return nil +} + +// preClean wipes what the plan says to wipe before a dataset materializes. +// Which directories a kind clears — and, for packs-gs, which it deliberately +// keeps — is a property of the kind, so the plan owns the list and a dry run +// prints exactly the wipes the run performs. +func preClean(s plan.Step, out io.Writer) error { + if len(s.PreClean) == 0 { + return nil + } + return removeAll(out, s.PreClean...) +} + +// runDatasetCommands runs the step's commands in order, stopping at the first +// failure: what they build together is one pack tree, and half of one is worth +// nothing. +func runDatasetCommands(s plan.Step, out io.Writer) error { + for _, argv := range s.Argv { + if err := runCommand(argv, s.Env, out); err != nil { + return err + } + } + return nil +} + +// goldenPresent reports whether a pack root is there and non-empty, the port of +// bash's golden_present. +func goldenPresent(dir string) bool { + f, err := os.Open(dir) + if err != nil { + return false + } + defer f.Close() + names, err := f.Readdirnames(1) + return err == nil && len(names) > 0 +} + +func dirExists(path string) bool { + fi, err := os.Stat(path) + return err == nil && fi.IsDir() +} + +// mkdirAll and rename log themselves as the commands bash ran, so a campaign +// log shows every filesystem move the runner made, not just the ones that +// happened to be external processes. +func mkdirAll(out io.Writer, dir string) error { + fmt.Fprintf(out, " $ mkdir -p %s\n", dir) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("mkdir -p %s: %w", dir, err) + } + return nil +} + +func rename(out io.Writer, from, to string) error { + fmt.Fprintf(out, " $ mv %s %s\n", from, to) + if err := os.Rename(from, to); err != nil { + return fmt.Errorf("mv %s %s: %w", from, to, err) + } + return nil +} diff --git a/runner/internal/run/dataset_test.go b/runner/internal/run/dataset_test.go new file mode 100644 index 0000000..672991f --- /dev/null +++ b/runner/internal/run/dataset_test.go @@ -0,0 +1,248 @@ +package run + +import ( + "path/filepath" + "strings" + "testing" + + "github.com/stellar/stellar-rpc-benchmarks/runner/internal/config" + "github.com/stellar/stellar-rpc-benchmarks/runner/internal/plan" +) + +// dsStep is a dataset step whose commands are shell scripts standing in for +// gcloud rsync and the bench binary: they write into .partial exactly +// where the real tools do, which is all the choreography under test cares +// about. PreClean mirrors what plan.Build populates for the kind, since that +// list — not the executor — is what decides which directories get wiped. +func dsStep(name, kind, root string, argv ...[]string) plan.Step { + if argv == nil { + argv = [][]string{} + } + s := plan.Step{ + ID: "dataset-" + name, + Kind: plan.KindDataset, + Argv: argv, + Dataset: &plan.DatasetSpec{Name: name, Kind: kind, Location: "gs://bucket/cold", Root: root}, + } + switch kind { + case config.KindPacksGS: + s.PreClean = []string{root} + case config.KindBSBS3: + s.PreClean = []string{root, root + ".partial"} + } + return s +} + +// fixtureStep is dsStep for the one kind that also owns a staging pack tree, +// which the plan wipes by its parent along with both roots. +func fixtureStep(name, root, stage string, argv ...[]string) plan.Step { + s := dsStep(name, config.KindFixture, root, argv...) + s.Dataset.Stage = stage + s.PreClean = []string{filepath.Dir(stage), root, root + ".partial"} + return s +} + +// materialize is a command that fills dir with a pack tree, the way a fetch, a +// backfill, or a freeze would. +func materialize(dir string) []string { + return []string{"/bin/sh", "-c", `mkdir -p "$1/ledgers" && : > "$1/ledgers/chunk-1.pack"`, "sh", dir} +} + +// marker is a command that records that it ran. Preparations that should have +// been short-circuited are proven by its absence. +func marker(path string) []string { + return []string{"/bin/sh", "-c", `: > "$1"`, "sh", path} +} + +func prepare(t *testing.T, step plan.Step) outcome { + t.Helper() + return walk(t, &plan.Plan{Steps: []plan.Step{step}}, Options{}) +} + +func TestPrepareDatasetPacksLocal(t *testing.T) { + t.Run("a root holding packs is accepted as it is", func(t *testing.T) { + root := t.TempDir() + mustWrite(t, filepath.Join(root, "ledgers", "chunk-1.pack"), "packs") + got := prepare(t, dsStep("local", config.KindPacksLocal, root)) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "dataset local: local cold pack root "+root) + }) + + t.Run("a root without ledgers/ is refused in the operator's own words", func(t *testing.T) { + root := t.TempDir() + got := prepare(t, dsStep("local", config.KindPacksLocal, root)) + got.assertStatuses(t, StatusFailed) + want := "dataset 'local': " + root + "/ledgers not found — location must be a cold pack root" + if err := got.results[0].Err; err == nil || err.Error() != want { + t.Errorf("error = %v, want %q", err, want) + } + }) +} + +func TestPrepareDatasetPacksGS(t *testing.T) { + t.Run("an empty leftover root is cleared and the partial renamed onto it", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "pubnet") + mustMkdir(t, root) // the leftover of an earlier, cleared-out fetch + // A half-fetched tree from a killed session: the partial is deliberately + // kept, because rsync resumes into it. + mustWrite(t, root+".partial/half.pack", "resumable") + + got := prepare(t, dsStep("pubnet", config.KindPacksGS, root, materialize(root+".partial"))) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "dataset pubnet: fetch gs://bucket/cold") + got.assertLogHas(t, " $ rm -rf "+root+"\n") + got.assertLogHas(t, " $ mkdir -p "+root+".partial") + got.assertLogHas(t, " $ mv "+root+".partial "+root) + + assertExists(t, filepath.Join(root, "ledgers", "chunk-1.pack")) + assertExists(t, filepath.Join(root, "half.pack")) // the resumed bytes + assertGone(t, root+".partial") + assertGone(t, filepath.Join(root, "pubnet.partial")) // never nested + }) + + t.Run("golden packs already there short-circuit the fetch", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "pubnet") + mustWrite(t, filepath.Join(root, "ledgers", "chunk-1.pack"), "packs") + ran := filepath.Join(tmp, "ran.txt") + + got := prepare(t, dsStep("pubnet", config.KindPacksGS, root, marker(ran))) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "dataset pubnet: golden packs already at "+root+" — skipping fetch") + assertGone(t, ran) + // The short-circuit comes before the wipes: present golden packs are + // never touched, whatever pre_clean says. + got.assertLogLacks(t, "rm -rf") + assertExists(t, filepath.Join(root, "ledgers", "chunk-1.pack")) + }) + + t.Run("a preparation that leaves no ledgers/ fails", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "pubnet") + got := prepare(t, dsStep("pubnet", config.KindPacksGS, root, marker(root+".partial/events.db"))) + got.assertStatuses(t, StatusFailed) + want := "dataset 'pubnet': " + root + "/ledgers missing after preparation" + if err := got.results[0].Err; err == nil || err.Error() != want { + t.Errorf("error = %v, want %q", err, want) + } + }) + + t.Run("a failed fetch leaves the partial behind and never renames it", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "pubnet") + got := prepare(t, dsStep("pubnet", config.KindPacksGS, root, []string{"/bin/sh", "-c", "exit 1"})) + got.assertStatuses(t, StatusFailed) + assertGone(t, root) + assertExists(t, root+".partial") + }) +} + +func TestPrepareDatasetBSBS3(t *testing.T) { + t.Run("a stale partial is wiped before the backfill", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "bsb") + stale := root + ".partial/stale.pack" + mustWrite(t, stale, "from a backfill that died") + + // The backfill only materializes when the stale bytes are gone: a + // resumed cold backfill would double-write the pack tree. + argv := []string{"/bin/sh", "-c", `test ! -e "$2" && mkdir -p "$1/ledgers"`, "sh", root + ".partial", stale} + got := prepare(t, dsStep("bsb", config.KindBSBS3, root, argv)) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, " $ rm -rf "+root+" "+root+".partial") + assertExists(t, filepath.Join(root, "ledgers")) + assertGone(t, filepath.Join(root, "stale.pack")) + }) + + t.Run("golden packs already there short-circuit the backfill", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "bsb") + mustWrite(t, filepath.Join(root, "ledgers", "chunk-1.pack"), "packs") + ran := filepath.Join(tmp, "ran.txt") + + got := prepare(t, dsStep("bsb", config.KindBSBS3, root, marker(ran))) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "dataset bsb: golden packs already at "+root+" — skipping backfill") + assertGone(t, ran) + }) + + t.Run("the S3 env is set on the backfill commands", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "bsb") + argv := []string{"/bin/sh", "-c", `test "$AWS_EC2_METADATA_DISABLED" = true && mkdir -p "$1/ledgers"`, "sh", root + ".partial"} + step := dsStep("bsb", config.KindBSBS3, root, argv) + step.Env = map[string]string{"AWS_EC2_METADATA_DISABLED": "true"} + + got := prepare(t, step) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, " $ env AWS_EC2_METADATA_DISABLED=true /bin/sh -c") + }) +} + +func TestPrepareDatasetFixture(t *testing.T) { + t.Run("a stale staging tree is wiped before generation", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "fix") + stage := filepath.Join(tmp, "fixture", "fix", "ledgers") + stale := filepath.Join(tmp, "fixture", "fix", "ledgers", "chunk-1.pack") + mustWrite(t, stale, "half a chunk from a killed generation") + + // Generation refuses to run on top of the stale chunk; the freeze then + // fills the partial. + generate := []string{"/bin/sh", "-c", `test ! -e "$2" && mkdir -p "$1"`, "sh", stage, stale} + freeze := materialize(root + ".partial") + got := prepare(t, fixtureStep("fix", root, stage, generate, freeze)) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "dataset fix: generate a fixture pack tree") + got.assertLogHas(t, " $ rm -rf "+filepath.Join(tmp, "fixture", "fix")+" "+root+" "+root+".partial") + assertExists(t, filepath.Join(root, "ledgers", "chunk-1.pack")) + }) + + t.Run("golden packs already there short-circuit the generation", func(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "fix") + mustWrite(t, filepath.Join(root, "ledgers", "chunk-1.pack"), "packs") + ran := filepath.Join(tmp, "ran.txt") + stage := filepath.Join(tmp, "fixture", "fix", "ledgers") + + staged := filepath.Join(stage, "chunk-1.pack") + mustWrite(t, staged, "left by the generation that made these packs") + + got := prepare(t, fixtureStep("fix", root, stage, marker(ran))) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "dataset fix: golden packs already at "+root+" — skipping generation") + assertGone(t, ran) + assertExists(t, staged) // the short-circuit precedes the staging wipe + }) +} + +// TestPrepareDatasetWipesWhatThePlanSays pins where the wipe list lives: the +// executor removes the step's pre_clean and nothing else, so a plan that asks +// for the partial gets it removed even for the kind that normally resumes. +func TestPrepareDatasetWipesWhatThePlanSays(t *testing.T) { + tmp := t.TempDir() + root := filepath.Join(tmp, "golden", "pubnet") + mustWrite(t, root+".partial/half.pack", "resumable, but this plan says otherwise") + + step := dsStep("pubnet", config.KindPacksGS, root, materialize(root+".partial")) + step.PreClean = []string{root, root + ".partial"} + + got := prepare(t, step) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, " $ rm -rf "+root+" "+root+".partial") + assertGone(t, filepath.Join(root, "half.pack")) +} + +func TestPrepareDatasetRejectsAnUnknownKind(t *testing.T) { + got := prepare(t, dsStep("odd", "packs-ftp", filepath.Join(t.TempDir(), "golden", "odd"))) + got.assertStatuses(t, StatusFailed) + if err := got.results[0].Err; err == nil || !strings.Contains(err.Error(), `unknown kind "packs-ftp"`) { + t.Errorf("error = %v, want it to name the unknown kind", err) + } +} + +func TestRunDatasetWithoutASpec(t *testing.T) { + got := prepare(t, plan.Step{ID: "dataset-x", Kind: plan.KindDataset, Argv: [][]string{}}) + got.assertStatuses(t, StatusFailed) +} diff --git a/runner/internal/run/doc.go b/runner/internal/run/doc.go new file mode 100644 index 0000000..b629bfd --- /dev/null +++ b/runner/internal/run/doc.go @@ -0,0 +1,3 @@ +// Package run executes a plan: it walks the steps, writes the per-leg +// completion sentinels, and decides what a resumed campaign may skip. +package run diff --git a/runner/internal/run/lock.go b/runner/internal/run/lock.go new file mode 100644 index 0000000..db788ae --- /dev/null +++ b/runner/internal/run/lock.go @@ -0,0 +1,39 @@ +package run + +import ( + "fmt" + "os" + "path/filepath" + "syscall" +) + +// lockName is the lock file at the root of a BENCH_ROOT. It is created once +// and never deleted: unlinking a lock file races with the next campaign, which +// may already hold the old inode open. +const lockName = ".campaign.lock" + +// AcquireLock takes an exclusive, non-blocking flock on /.campaign.lock +// and returns the function that releases it. Two campaigns sharing a BENCH_ROOT +// would fight over the same build clone, scratch dirs, and hot DBs, so the +// second one is refused immediately rather than queued: the operator wants to +// know now, not in six hours. +func AcquireLock(benchRoot string) (release func(), err error) { + if err := os.MkdirAll(benchRoot, 0o755); err != nil { + return nil, fmt.Errorf("create BENCH_ROOT %s: %w", benchRoot, err) + } + path := filepath.Join(benchRoot, lockName) + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, fmt.Errorf("open lock file %s: %w", path, err) + } + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + f.Close() + return nil, fmt.Errorf("another campaign is already running on this BENCH_ROOT (held lock: %s)", path) + } + // The lock lives on the open file description, so the fd stays open until + // release; closing it anywhere earlier would drop the lock silently. + return func() { + syscall.Flock(int(f.Fd()), syscall.LOCK_UN) + f.Close() + }, nil +} diff --git a/runner/internal/run/log.go b/runner/internal/run/log.go new file mode 100644 index 0000000..618a6e8 --- /dev/null +++ b/runner/internal/run/log.go @@ -0,0 +1,26 @@ +package run + +import ( + "fmt" + "io" + "os" + "path/filepath" + "time" +) + +// campaignLogName is the per-bundle log every session appends to. +const campaignLogName = "campaign.log" + +// Notef prints a bash-style note — `== [HH:MM:SS] msg`, the clock in UTC — +// the line format every operator reading a campaign log already knows from +// campaign.sh's note(). +func Notef(w io.Writer, format string, args ...any) { + fmt.Fprintf(w, "== [%s] %s\n", time.Now().UTC().Format("15:04:05"), fmt.Sprintf(format, args...)) +} + +// OpenCampaignLog opens /campaign.log for appending. Append, not +// truncate: a campaign that is resumed twice leaves all three sessions in one +// file, in the order they happened. +func OpenCampaignLog(resultsDir string) (*os.File, error) { + return os.OpenFile(filepath.Join(resultsDir, campaignLogName), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) +} diff --git a/runner/internal/run/resume.go b/runner/internal/run/resume.go new file mode 100644 index 0000000..94f6f86 --- /dev/null +++ b/runner/internal/run/resume.go @@ -0,0 +1,140 @@ +package run + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" +) + +// legStateKind is what an existing --out directory means to a resumed campaign. +type legStateKind int + +const ( + legAbsent legStateKind = iota // no directory at all + legPartial // something is there, but nothing says it finished + legFailedEarlier // a previous session ran this leg and it failed + legComplete // a previous session ran this leg and it succeeded +) + +// legState is a classification plus, for a failure, the reason to show the +// operator — the sentinel's error, the exit status, or the invocation manifest's +// own error field. +type legState struct { + kind legStateKind + reason string +} + +// legSentinelName is the runner-owned completion marker. The bench subcommands +// own invocation.json; nothing but this runner writes leg.json, which is why it +// is the sentinel resume trusts first. +const legSentinelName = "leg.json" + +// LegComplete reports whether a leg's --out directory already holds a leg an +// earlier session finished successfully. wantID is the plan's id for the leg, +// which the sentinel must name to be trusted. It is the read-only half of the +// resume decision: `run --dry-run --resume` annotates the plan with it, +// touching nothing. +func LegComplete(dir, wantID string) bool { + return classifyLegDir(dir, wantID).kind == legComplete +} + +// classifyLegDir decides what a resumed campaign should do with a leg's +// existing --out directory. Only a marker that positively records success +// counts as complete; every ambiguous state resolves to "wipe and re-run", +// because a half-written leg silently kept would corrupt the aggregates the +// converter computes over the bundle. +func classifyLegDir(dir, wantID string) legState { + // Only positive absence is absence; everything unreadable resolves to + // wipe-and-re-run. Lstat rather than Stat so a dangling symlink is seen as + // something-is-there, and permission or I/O errors fall to partial too — + // otherwise resume would call MkdirAll on a path that can never be created. + if _, err := os.Lstat(dir); err != nil { + if os.IsNotExist(err) { + return legState{kind: legAbsent} + } + return legState{kind: legPartial} + } + + // A sentinel is trusted only once it has identified itself: this runner's + // schema version and this leg's id. A `{}` that happens to parse, a record + // copied in from another leg, or a future schema whose fields mean something + // else all prove nothing about this directory, so they are partial. + switch sentinel, err := readLegSentinel(filepath.Join(dir, legSentinelName)); { + case err == nil && (sentinel.SchemaVersion != LegSchemaVersion || sentinel.ID != wantID): + return legState{kind: legPartial, reason: "sentinel does not match this leg"} + case err == nil && sentinel.ExitCode == nil: + return legState{kind: legPartial, reason: "sentinel records no exit"} + case err == nil && *sentinel.ExitCode == 0 && sentinel.Error == "": + return legState{kind: legComplete} + case err == nil && sentinel.Error != "": + return legState{kind: legFailedEarlier, reason: sentinel.Error} + case err == nil: + return legState{kind: legFailedEarlier, reason: fmt.Sprintf("exit status %d", *sentinel.ExitCode)} + case !os.IsNotExist(err): + // The sentinel is there but unreadable or corrupt: it proves nothing, + // so the leg is treated as partial rather than trusted either way. + return legState{kind: legPartial} + } + + // No leg.json: this may be a bundle the bash runner produced, which had no + // sentinel of its own and inferred completion from the manifests the bench + // subcommand writes. A failed run also writes invocation.json — with an + // `error` field (stellar-rpc#907) — so completion means both files present + // AND no error recorded. + inv, err := readInvocation(filepath.Join(dir, "invocation.json")) + if err != nil { + return legState{kind: legPartial} + } + if _, err := os.Stat(filepath.Join(dir, "driver.csv")); err != nil { + return legState{kind: legPartial} + } + if inv.Error != "" { + return legState{kind: legFailedEarlier, reason: inv.Error} + } + return legState{kind: legComplete} +} + +// legSentinelView is the read side of leg.json: the fields resume decides on. +// ExitCode is a pointer because "the runner recorded exit 0" and "there is no +// exit_code here at all" must not read alike — decoding an absent field into an +// int would turn a sentinel that records no exit into a claim of success. +type legSentinelView struct { + SchemaVersion int `json:"schema_version"` + ID string `json:"id"` + ExitCode *int `json:"exit_code"` + Error string `json:"error"` +} + +// readLegSentinel reads and parses a leg.json. A missing file is reported as +// os.IsNotExist so the caller can fall back to the bash-era manifests. +func readLegSentinel(path string) (legSentinelView, error) { + b, err := os.ReadFile(path) + if err != nil { + return legSentinelView{}, err + } + var s legSentinelView + if err := json.Unmarshal(b, &s); err != nil { + return legSentinelView{}, err + } + return s, nil +} + +// invocationManifest is the sliver of stellar-rpc's invocation.json this runner +// reads: whether the run recorded a failure. The file is camelCase and owned by +// the other repo; the runner never writes it. +type invocationManifest struct { + Error string `json:"error"` +} + +func readInvocation(path string) (invocationManifest, error) { + b, err := os.ReadFile(path) + if err != nil { + return invocationManifest{}, err + } + var m invocationManifest + if err := json.Unmarshal(b, &m); err != nil { + return invocationManifest{}, err + } + return m, nil +} diff --git a/runner/internal/run/run.go b/runner/internal/run/run.go new file mode 100644 index 0000000..0499571 --- /dev/null +++ b/runner/internal/run/run.go @@ -0,0 +1,376 @@ +package run + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/stellar/stellar-rpc-benchmarks/runner/internal/plan" +) + +// LegSchemaVersion is the version of the leg.json contract. Like plan.json, +// additive changes keep the version. +const LegSchemaVersion = 1 + +// Status is what became of one step. +type Status string + +const ( + StatusOK Status = "ok" + StatusFailed Status = "failed" + StatusSkipped Status = "skipped" // a need (transitively) failed + StatusResumed Status = "resumed" // already complete in an earlier session +) + +// StepResult is one line of the campaign's outcome. +type StepResult struct { + ID string + Status Status + Err error // nil unless failed +} + +// Options configures a walk. +type Options struct { + // Output receives the runner's notes, the command lines, and the child + // processes' stdout and stderr. The caller composes the tee (terminal plus + // campaign.log); the executor just writes. Defaults to os.Stdout, which is + // where bash sent everything. + Output io.Writer + // Resume inspects each leg's existing --out directory and skips the ones an + // earlier session finished. Off, nothing existing is read or wiped by the + // resume path. + Resume bool + // FailFast stops the walk at the first failed step. Default (keep going): a + // failure only skips the steps that need it. + FailFast bool + // OnStepDone is called after every step the walk executed — including the + // ones a resume or an existing binary made a no-op, excluding only the ones + // skipped because a need failed. The run wiring writes binary.txt from it + // the moment the build succeeds, so a campaign that dies during its legs + // still leaves the binary's identity in the bundle. + OnStepDone func(s plan.Step, res StepResult) +} + +// legSentinel is leg.json: the runner's own record that a leg ran to +// completion, written whether it succeeded or not. The bench subcommands' +// invocation.json cannot play this role — it is written by the process being +// measured, so a process killed before it got there leaves no trace at all. +// +// ExitCode carries no omitempty on purpose: exit 0 is the success record resume +// reads, and a sentinel without the field is not trusted (see legSentinelView). +type legSentinel struct { + SchemaVersion int `json:"schema_version"` + ID string `json:"id"` + Argv []string `json:"argv"` + ExitCode int `json:"exit_code"` + StartedAt string `json:"started_at"` + FinishedAt string `json:"finished_at"` + DurationNS int64 `json:"duration_ns"` + Error string `json:"error,omitempty"` +} + +// Execute walks the plan in order — sequentially, always: these are benchmarks, +// and two of them sharing the machine measure the sharing. It returns one +// result per executed step and a non-nil error when any step failed; the +// failure summary has already been printed to Output, so a caller that exits +// nonzero needs to print nothing more. +// +// Under FailFast the walk stops at the first failure, so the returned slice is +// short: steps after the failed one have no result at all, rather than a +// skipped one. +func Execute(p *plan.Plan, opts Options) ([]StepResult, error) { + if opts.Output == nil { + opts.Output = os.Stdout + } + results := make([]StepResult, 0, len(p.Steps)) + // bad holds every step that failed or was skipped. Skipped steps being bad + // is what makes the propagation transitive: the dependent of a skipped step + // is skipped too, without walking the graph. + bad := map[string]bool{} + skipNeed := map[string]string{} + + for _, step := range p.Steps { + // The tarball and the publish are the wiring's epilogue, not the walk's: + // tar must run after the final provenance writes so the bundle it + // preserves contains them, and a publish failure is not a benchmark + // failure. They are in the plan because they are part of the campaign; + // they are skipped here because they belong after every step of it. + if step.Kind == plan.KindTarball || step.Kind == plan.KindPublish { + continue + } + if need := firstBadNeed(step, bad); need != "" { + Notef(opts.Output, "skipping %s: needs %s, which failed or was skipped", step.ID, need) + bad[step.ID] = true + skipNeed[step.ID] = need + results = append(results, StepResult{ID: step.ID, Status: StatusSkipped}) + continue + } + Notef(opts.Output, "%s", step.ID) + res := executeStep(p, step, opts) + results = append(results, res) + if opts.OnStepDone != nil { + opts.OnStepDone(step, res) + } + if res.Status == StatusFailed { + Notef(opts.Output, "%s failed: %v", step.ID, res.Err) + bad[step.ID] = true + if opts.FailFast { + break + } + } + } + return results, summarize(results, skipNeed, opts.Output) +} + +// firstBadNeed returns the first of a step's needs that failed or was skipped, +// or "" when the step is clear to run. +func firstBadNeed(s plan.Step, bad map[string]bool) string { + for _, need := range s.Needs { + if bad[need] { + return need + } + } + return "" +} + +// summarize prints the end-of-campaign failure block and returns the error +// Execute hands back. An all-ok campaign prints nothing and returns nil. +func summarize(results []StepResult, skipNeed map[string]string, w io.Writer) error { + var failed, skipped []StepResult + for _, r := range results { + switch r.Status { + case StatusFailed: + failed = append(failed, r) + case StatusSkipped: + skipped = append(skipped, r) + } + } + if len(failed) == 0 && len(skipped) == 0 { + return nil + } + fmt.Fprintf(w, "== campaign summary: %d failed, %d skipped\n", len(failed), len(skipped)) + for _, r := range failed { + fmt.Fprintf(w, "== failed: %s (%v)\n", r.ID, r.Err) + } + for _, r := range skipped { + fmt.Fprintf(w, "== skipped: %s (needs %s)\n", r.ID, skipNeed[r.ID]) + } + if len(failed) == 0 { + return nil + } + return fmt.Errorf("%d step(s) failed", len(failed)) +} + +func executeStep(p *plan.Plan, s plan.Step, opts Options) StepResult { + switch s.Kind { + case plan.KindLeg: + return runLeg(s, opts) + case plan.KindBuild: + return runBuild(p, s, opts) + case plan.KindDataset: + return runDataset(s, opts) + default: + return failure(s, fmt.Errorf("unknown step kind %q", s.Kind)) + } +} + +// runLeg runs one timed benchmark leg: the whole point of the campaign, and the +// only step kind with a completion sentinel. +func runLeg(s plan.Step, opts Options) StepResult { + if len(s.Argv) != 1 { + return failure(s, fmt.Errorf("leg has %d commands, want exactly 1 (the measurement is the process)", len(s.Argv))) + } + if opts.Resume { + base := filepath.Base(s.OutDir) + state := classifyLegDir(s.OutDir, s.ID) + switch { + case state.kind == legComplete: + Notef(opts.Output, "resume: %s already complete — skipping", base) + return StepResult{ID: s.ID, Status: StatusResumed} + case state.kind == legFailedEarlier: + Notef(opts.Output, "resume: %s failed in an earlier session (%s) — wiping and re-running", base, state.reason) + case state.kind == legPartial && state.reason != "": + Notef(opts.Output, "resume: %s is a partial leg (%s) — wiping and re-running", base, state.reason) + case state.kind == legPartial: + Notef(opts.Output, "resume: %s is a partial leg — wiping and re-running", base) + } + if state.kind != legAbsent { + if err := removeAll(opts.Output, s.OutDir); err != nil { + return failure(s, err) + } + } + } + for _, dir := range s.PreClean { + if err := removeAll(opts.Output, dir); err != nil { + return failure(s, err) + } + } + + // The bench subcommand creates its own --out dir, but creating it here too + // means the sentinel has somewhere to land even when the binary dies + // instantly — which is exactly the case resume most needs to classify. + started := time.Now() + runErr := os.MkdirAll(s.OutDir, 0o755) + if runErr == nil { + runErr = runCommand(s.Argv[0], s.Env, opts.Output) + } + finished := time.Now() + + if err := writeLegSentinel(s, started, finished, runErr); err != nil { + if runErr == nil { + // A leg whose completion cannot be recorded is not complete: a + // resume would re-run it anyway, so call it failed now. + runErr = err + } else { + Notef(opts.Output, "warning: %s: %v", s.ID, err) + } + } + if runErr != nil { + return failure(s, runErr) + } + // Post-cleaning only on success keeps a failed leg's scratch around for + // diagnosis. A failure to clean is not a failure of the measurement. + for _, dir := range s.PostClean { + if err := removeAll(opts.Output, dir); err != nil { + Notef(opts.Output, "warning: %s: %v", s.ID, err) + } + } + return StepResult{ID: s.ID, Status: StatusOK} +} + +// writeLegSentinel records how the leg went, success or failure, into its --out +// directory. +func writeLegSentinel(s plan.Step, started, finished time.Time, runErr error) error { + sentinel := legSentinel{ + SchemaVersion: LegSchemaVersion, + ID: s.ID, + Argv: s.Argv[0], + StartedAt: started.UTC().Format(time.RFC3339), + FinishedAt: finished.UTC().Format(time.RFC3339), + DurationNS: finished.Sub(started).Nanoseconds(), + } + if runErr != nil { + sentinel.ExitCode = exitCode(runErr) + sentinel.Error = runErr.Error() + } + b, err := json.MarshalIndent(sentinel, "", " ") + if err != nil { + return fmt.Errorf("marshal %s: %w", legSentinelName, err) + } + if err := os.WriteFile(filepath.Join(s.OutDir, legSentinelName), append(b, '\n'), 0o644); err != nil { + return fmt.Errorf("write %s: %w", legSentinelName, err) + } + return nil +} + +// exitCode is the child's status, or -1 when it never got far enough to have +// one (binary missing, permission denied, signal). +func exitCode(err error) int { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { + return exitErr.ExitCode() + } + return -1 +} + +// runBuild builds the binary under test unless it is already there. The binary +// at its versioned path is its own completion marker — the path contains the +// commit's short sha, so a stale binary cannot be mistaken for this one. +func runBuild(p *plan.Plan, s plan.Step, opts Options) StepResult { + if executableExists(p.Bin) { + Notef(opts.Output, "binary %s already built — skipping build", p.Bin) + return StepResult{ID: s.ID, Status: StatusOK} + } + return runCommands(s, opts) +} + +// executableExists reports whether path is a regular file anyone may execute. +func executableExists(path string) bool { + fi, err := os.Stat(path) + return err == nil && fi.Mode().IsRegular() && fi.Mode().Perm()&0o111 != 0 +} + +// runCommands runs a step's commands in order, stopping at the first failure. +// They belong to one step precisely because stopping between them would leave +// something half-made. +func runCommands(s plan.Step, opts Options) StepResult { + for _, argv := range s.Argv { + if err := runCommand(argv, s.Env, opts.Output); err != nil { + return failure(s, err) + } + } + return StepResult{ID: s.ID, Status: StatusOK} +} + +// RunStep runs one step's commands outside the walk, printed and plumbed +// exactly as Execute would. It exists for the steps Execute deliberately leaves +// to the wiring's epilogue — the tarball, which must be made after the final +// provenance writes. +func RunStep(s plan.Step, out io.Writer) error { + if res := runCommands(s, Options{Output: out}); res.Status != StatusOK { + return res.Err + } + return nil +} + +// runCommand prints the command the way the plan printer and bash's run() do, +// then executes it with its output going wherever the notes go. +func runCommand(argv []string, env map[string]string, out io.Writer) error { + if len(argv) == 0 { + return errors.New("empty command") + } + fmt.Fprintf(out, " $ %s%s\n", envPrefix(env), strings.Join(argv, " ")) + cmd := exec.Command(argv[0], argv[1:]...) + cmd.Stdout, cmd.Stderr = out, out + cmd.Env = os.Environ() + for _, k := range sortedKeys(env) { + cmd.Env = append(cmd.Env, k+"="+env[k]) + } + return cmd.Run() +} + +// removeAll wipes directories, logging them as the single rm -rf bash ran. +func removeAll(out io.Writer, dirs ...string) error { + fmt.Fprintf(out, " $ rm -rf %s\n", strings.Join(dirs, " ")) + for _, dir := range dirs { + if err := os.RemoveAll(dir); err != nil { + return fmt.Errorf("rm -rf %s: %w", dir, err) + } + } + return nil +} + +// envPrefix renders a step's extra environment as an `env K=V ` command prefix, +// matching plan.Plan.Print so a log line and a plan line for the same command +// read identically. +func envPrefix(env map[string]string) string { + if len(env) == 0 { + return "" + } + var b strings.Builder + b.WriteString("env ") + for _, k := range sortedKeys(env) { + fmt.Fprintf(&b, "%s=%s ", k, env[k]) + } + return b.String() +} + +func sortedKeys(env map[string]string) []string { + keys := make([]string, 0, len(env)) + for k := range env { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +func failure(s plan.Step, err error) StepResult { + return StepResult{ID: s.ID, Status: StatusFailed, Err: err} +} diff --git a/runner/internal/run/run_test.go b/runner/internal/run/run_test.go new file mode 100644 index 0000000..67c2816 --- /dev/null +++ b/runner/internal/run/run_test.go @@ -0,0 +1,707 @@ +package run + +import ( + "bytes" + "encoding/json" + "io" + "os" + "path/filepath" + "regexp" + "slices" + "strings" + "testing" + + "github.com/stellar/stellar-rpc-benchmarks/runner/internal/plan" +) + +// --- helpers -------------------------------------------------------------- + +func mustMkdir(t *testing.T, dir string) { + t.Helper() + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", dir, err) + } +} + +func mustWrite(t *testing.T, path, body string) { + t.Helper() + mustMkdir(t, filepath.Dir(path)) + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatalf("write %s: %v", path, err) + } +} + +// shLeg is a timed leg whose measurement is a shell script. The script's +// positional parameters carry the paths it needs — passing them as arguments +// rather than environment keeps the leg's Env free for the env-propagation +// test. +func shLeg(id, outDir, script string, args ...string) plan.Step { + argv := append([]string{"/bin/sh", "-c", script, "sh", outDir}, args...) + return plan.Step{ID: id, Kind: plan.KindLeg, Timed: true, OutDir: outDir, Argv: [][]string{argv}} +} + +// outcome is everything one Execute call produced. +type outcome struct { + results []StepResult + err error + log string +} + +func walk(t *testing.T, p *plan.Plan, opts Options) outcome { + t.Helper() + var buf bytes.Buffer + opts.Output = &buf + results, err := Execute(p, opts) + return outcome{results: results, err: err, log: buf.String()} +} + +func (o outcome) statuses() []Status { + got := make([]Status, len(o.results)) + for i, r := range o.results { + got[i] = r.Status + } + return got +} + +func (o outcome) assertStatuses(t *testing.T, want ...Status) { + t.Helper() + if got := o.statuses(); !slices.Equal(got, want) { + t.Errorf("statuses = %v, want %v\nlog:\n%s", got, want, o.log) + } +} + +func (o outcome) assertLogHas(t *testing.T, want string) { + t.Helper() + if !strings.Contains(o.log, want) { + t.Errorf("log does not contain %q\nlog:\n%s", want, o.log) + } +} + +func (o outcome) assertLogLacks(t *testing.T, unwanted string) { + t.Helper() + if strings.Contains(o.log, unwanted) { + t.Errorf("log contains %q, want it not to\nlog:\n%s", unwanted, o.log) + } +} + +func readSentinel(t *testing.T, outDir string) legSentinel { + t.Helper() + path := filepath.Join(outDir, legSentinelName) + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + var s legSentinel + if err := json.Unmarshal(b, &s); err != nil { + t.Fatalf("unmarshal %s: %v", path, err) + } + return s +} + +func assertGone(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("%s still exists (stat err = %v)", path, err) + } +} + +func assertExists(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); err != nil { + t.Errorf("%s missing: %v", path, err) + } +} + +// --- resume decision table ------------------------------------------------- + +func TestClassifyLegDir(t *testing.T) { + // The leg being classified: its plan id is its --out directory's basename, + // and the sentinel has to name it to be believed. + const legID = "ingest-cold-ds-c0-run1" + cases := []struct { + name string + setup func(t *testing.T, dir string) + want legStateKind + wantReason string + }{ + { + name: "no directory at all", + setup: func(t *testing.T, dir string) { os.RemoveAll(dir) }, + want: legAbsent, + }, + { + name: "empty directory", + setup: func(t *testing.T, dir string) {}, + want: legPartial, + }, + { + name: "dangling symlink where the out-dir should be", + setup: func(t *testing.T, dir string) { + if err := os.RemoveAll(dir); err != nil { + t.Fatalf("RemoveAll: %v", err) + } + if err := os.Symlink(dir+"-gone", dir); err != nil { + t.Fatalf("Symlink: %v", err) + } + }, + want: legPartial, + }, + { + name: "sentinel says success", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), `{"schema_version":1,"id":"`+legID+`","exit_code":0}`) + }, + want: legComplete, + }, + { + name: "sentinel says failure", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), + `{"schema_version":1,"id":"`+legID+`","exit_code":1,"error":"exit status 1"}`) + }, + want: legFailedEarlier, + wantReason: "exit status 1", + }, + { + name: "sentinel with a nonzero exit and no error field", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), `{"schema_version":1,"id":"`+legID+`","exit_code":2}`) + }, + want: legFailedEarlier, + wantReason: "exit status 2", + }, + { + // The degenerate sentinel: valid JSON, zero exit code by omission, + // and no claim to be anything. Believing it would skip a leg that + // never ran. + name: "empty JSON object as a sentinel", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), `{}`) + }, + want: legPartial, + wantReason: "sentinel does not match this leg", + }, + { + // The right leg, the right schema, and no record of how it ended: + // an absent exit_code must not decode into the 0 that means success. + name: "sentinel without an exit_code", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), `{"schema_version":1,"id":"`+legID+`"}`) + }, + want: legPartial, + wantReason: "sentinel records no exit", + }, + { + name: "sentinel records a different leg", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), + `{"schema_version":1,"id":"ingest-cold-ds-c0-run2","exit_code":0}`) + }, + want: legPartial, + wantReason: "sentinel does not match this leg", + }, + { + name: "sentinel from an unknown schema version", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), `{"schema_version":0,"id":"`+legID+`","exit_code":0}`) + }, + want: legPartial, + wantReason: "sentinel does not match this leg", + }, + { + name: "corrupt sentinel", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, legSentinelName), `{"schema_version":1,`) + }, + want: legPartial, + }, + { + name: "bash-era bundle: invocation.json and driver.csv, no error", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, "invocation.json"), + `{"schemaVersion":1,"command":"bench-ingest cold","startedAt":"2026-07-01T00:00:00Z","finishedAt":"2026-07-01T00:10:00Z"}`) + mustWrite(t, filepath.Join(dir, "driver.csv"), "stage,wall\n") + }, + want: legComplete, + }, + { + name: "bash-era bundle: invocation.json records an error", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, "invocation.json"), `{"schemaVersion":1,"error":"datastore unreachable"}`) + mustWrite(t, filepath.Join(dir, "driver.csv"), "stage,wall\n") + }, + want: legFailedEarlier, + wantReason: "datastore unreachable", + }, + { + name: "invocation.json without driver.csv", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, "invocation.json"), `{"schemaVersion":1}`) + }, + want: legPartial, + }, + { + name: "driver.csv without invocation.json", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, "driver.csv"), "stage,wall\n") + }, + want: legPartial, + }, + { + name: "unreadable invocation.json", + setup: func(t *testing.T, dir string) { + mustWrite(t, filepath.Join(dir, "invocation.json"), `{"schemaVersion":`) + mustWrite(t, filepath.Join(dir, "driver.csv"), "stage,wall\n") + }, + want: legPartial, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + dir := filepath.Join(t.TempDir(), legID) + mustMkdir(t, dir) + tc.setup(t, dir) + got := classifyLegDir(dir, legID) + if got.kind != tc.want { + t.Errorf("kind = %v, want %v", got.kind, tc.want) + } + if got.reason != tc.wantReason { + t.Errorf("reason = %q, want %q", got.reason, tc.wantReason) + } + }) + } +} + +// --- executor mechanics ----------------------------------------------------- + +func TestExecuteHappyPath(t *testing.T) { + tmp := t.TempDir() + scratch := filepath.Join(tmp, "scratch") + post := filepath.Join(tmp, "post") + mustWrite(t, filepath.Join(scratch, "stale"), "left over from the previous rep") + mustWrite(t, filepath.Join(post, "hot.db"), "x") + outA := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run1") + outB := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run2") + + a := shLeg("a", outA, `test ! -e "$2/stale" && : > "$1/driver.csv"`, scratch) + a.PreClean = []string{scratch} + a.PostClean = []string{post} + b := shLeg("b", outB, `: > "$1/driver.csv"`) + b.Needs = []string{"a"} + p := &plan.Plan{Steps: []plan.Step{a, b}} + + got := walk(t, p, Options{}) + if got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + got.assertStatuses(t, StatusOK, StatusOK) + + for _, out := range []string{outA, outB} { + assertExists(t, filepath.Join(out, "driver.csv")) + s := readSentinel(t, out) + if s.SchemaVersion != LegSchemaVersion || s.ExitCode != 0 || s.Error != "" { + t.Errorf("%s sentinel = %+v, want schema %d, exit 0, no error", out, s, LegSchemaVersion) + } + if s.DurationNS <= 0 { + t.Errorf("%s duration_ns = %d, want > 0", out, s.DurationNS) + } + if s.StartedAt == "" || s.FinishedAt == "" { + t.Errorf("%s sentinel is missing timestamps: %+v", out, s) + } + } + if s := readSentinel(t, outA); !slices.Equal(s.Argv, p.Steps[0].Argv[0]) { + t.Errorf("sentinel argv = %v, want %v", s.Argv, p.Steps[0].Argv[0]) + } + // PostClean runs only on success, and it ran: the hot DB is gone. + assertGone(t, post) + got.assertLogHas(t, " $ rm -rf "+scratch) +} + +func TestExecuteKeepGoing(t *testing.T) { + tmp := t.TempDir() + out := func(id string) string { return filepath.Join(tmp, "res", id) } + + fail := shLeg("fail", out("fail"), `exit 1`) + dep := shLeg("dep", out("dep"), `: > "$1/driver.csv"`) + dep.Needs = []string{"fail"} + dep2 := shLeg("dep2", out("dep2"), `: > "$1/driver.csv"`) + dep2.Needs = []string{"dep"} + indep := shLeg("indep", out("indep"), `: > "$1/driver.csv"`) + p := &plan.Plan{Steps: []plan.Step{fail, dep, dep2, indep}} + + got := walk(t, p, Options{}) + if got.err == nil { + t.Fatalf("Execute returned nil error after a failed leg\nlog:\n%s", got.log) + } + if want := "1 step(s) failed"; got.err.Error() != want { + t.Errorf("Execute error = %q, want %q", got.err, want) + } + got.assertStatuses(t, StatusFailed, StatusSkipped, StatusSkipped, StatusOK) + + got.assertLogHas(t, "== campaign summary: 1 failed, 2 skipped") + got.assertLogHas(t, "== failed: fail (exit status 1)") + got.assertLogHas(t, "== skipped: dep (needs fail)") + // dep2 needs dep, which was skipped rather than failed: the propagation is + // transitive without the executor walking the graph. + got.assertLogHas(t, "== skipped: dep2 (needs dep)") + + s := readSentinel(t, out("fail")) + if s.ExitCode != 1 || s.Error != "exit status 1" { + t.Errorf("failed leg sentinel = %+v, want exit_code 1 and error \"exit status 1\"", s) + } + assertExists(t, filepath.Join(out("indep"), legSentinelName)) + assertGone(t, out("dep")) +} + +func TestExecuteFailFast(t *testing.T) { + tmp := t.TempDir() + out := func(id string) string { return filepath.Join(tmp, "res", id) } + fail := shLeg("fail", out("fail"), `exit 3`) + indep := shLeg("indep", out("indep"), `: > "$1/driver.csv"`) + p := &plan.Plan{Steps: []plan.Step{fail, indep}} + + got := walk(t, p, Options{FailFast: true}) + if got.err == nil { + t.Fatalf("Execute returned nil error under fail-fast\nlog:\n%s", got.log) + } + got.assertStatuses(t, StatusFailed) + // The walk stopped: the later step has no result and never ran. + assertGone(t, out("indep")) + if s := readSentinel(t, out("fail")); s.ExitCode != 3 { + t.Errorf("sentinel exit_code = %d, want 3", s.ExitCode) + } +} + +func TestExecuteResume(t *testing.T) { + tmp := t.TempDir() + ran := filepath.Join(tmp, "ran.txt") + outDone := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run1") + outPartial := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run2") + + // An earlier session finished run1 and was killed during run2, which left a + // half-written directory with no manifests in it. + mustWrite(t, filepath.Join(outDone, legSentinelName), `{"schema_version":1,"id":"done","exit_code":0}`) + mustWrite(t, filepath.Join(outDone, "driver.csv"), "stage,wall\n") + mustWrite(t, filepath.Join(outPartial, "driver.csv.tmp"), "half a row") + + script := `echo "$2" >> "$3"; : > "$1/driver.csv"` + p := &plan.Plan{Steps: []plan.Step{ + shLeg("done", outDone, script, "done", ran), + shLeg("partial", outPartial, script, "partial", ran), + }} + + got := walk(t, p, Options{Resume: true}) + if got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + got.assertStatuses(t, StatusResumed, StatusOK) + got.assertLogHas(t, "resume: ingest-cold-ds-c0-run1 already complete — skipping") + got.assertLogHas(t, "resume: ingest-cold-ds-c0-run2 is a partial leg — wiping and re-running") + + b, err := os.ReadFile(ran) + if err != nil { + t.Fatalf("read %s: %v", ran, err) + } + if got := strings.Fields(string(b)); !slices.Equal(got, []string{"partial"}) { + t.Errorf("legs that ran = %v, want only [partial]", got) + } + // The partial directory was wiped, not merged into. + assertGone(t, filepath.Join(outPartial, "driver.csv.tmp")) + assertExists(t, filepath.Join(outPartial, legSentinelName)) +} + +func TestExecuteResumeAfterRecordedFailure(t *testing.T) { + tmp := t.TempDir() + out := filepath.Join(tmp, "res", "query-cold-ds-c0-run1") + mustWrite(t, filepath.Join(out, legSentinelName), `{"schema_version":1,"id":"leg","exit_code":1,"error":"exit status 1"}`) + + p := &plan.Plan{Steps: []plan.Step{shLeg("leg", out, `: > "$1/driver.csv"`)}} + got := walk(t, p, Options{Resume: true}) + if got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "resume: query-cold-ds-c0-run1 failed in an earlier session (exit status 1) — wiping and re-running") + if s := readSentinel(t, out); s.ExitCode != 0 || s.Error != "" { + t.Errorf("sentinel after re-run = %+v, want a clean success", s) + } +} + +// A sentinel that does not name this leg proves nothing about the directory it +// sits in — a copied or hand-made leg.json must not skip a leg that never ran. +func TestExecuteResumeRejectsForeignSentinel(t *testing.T) { + tmp := t.TempDir() + out := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run1") + mustWrite(t, filepath.Join(out, legSentinelName), `{"schema_version":1,"id":"ingest-cold-ds-c0-run2","exit_code":0}`) + ran := filepath.Join(tmp, "ran.txt") + + p := &plan.Plan{Steps: []plan.Step{shLeg("ingest-cold-ds-c0-run1", out, `echo ran >> "$2"; : > "$1/driver.csv"`, ran)}} + got := walk(t, p, Options{Resume: true}) + if got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "resume: ingest-cold-ds-c0-run1 is a partial leg (sentinel does not match this leg) — wiping and re-running") + assertExists(t, ran) + if s := readSentinel(t, out); s.ID != "ingest-cold-ds-c0-run1" { + t.Errorf("sentinel after re-run = %+v, want this leg's id", s) + } +} + +func TestExecuteWithoutResumeIgnoresExistingOutput(t *testing.T) { + tmp := t.TempDir() + out := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run1") + mustWrite(t, filepath.Join(out, legSentinelName), `{"schema_version":1,"exit_code":0}`) + ran := filepath.Join(tmp, "ran.txt") + + p := &plan.Plan{Steps: []plan.Step{shLeg("leg", out, `echo ran >> "$2"`, ran)}} + got := walk(t, p, Options{}) + got.assertStatuses(t, StatusOK) + assertExists(t, ran) + if strings.Contains(got.log, "resume:") { + t.Errorf("a non-resume walk inspected existing output\nlog:\n%s", got.log) + } +} + +func TestExecuteLegEnv(t *testing.T) { + tmp := t.TempDir() + out := filepath.Join(tmp, "res", "golden-ds-c0") + step := shLeg("env", out, `test "$FOO" = bar`) + step.Env = map[string]string{"FOO": "bar"} + + got := walk(t, &plan.Plan{Steps: []plan.Step{step}}, Options{}) + if got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, " $ env FOO=bar /bin/sh -c") +} + +func TestExecuteBuild(t *testing.T) { + t.Run("existing binary skips the build", func(t *testing.T) { + tmp := t.TempDir() + bin := filepath.Join(tmp, "bin", "stellar-rpc-deadbeef") + mustWrite(t, bin, "#!/bin/sh\n") + if err := os.Chmod(bin, 0o755); err != nil { + t.Fatalf("chmod: %v", err) + } + p := &plan.Plan{Bin: bin, Steps: []plan.Step{{ + ID: "build", Kind: plan.KindBuild, Argv: [][]string{{"/bin/sh", "-c", "exit 1"}}, + }}} + got := walk(t, p, Options{}) + got.assertStatuses(t, StatusOK) + got.assertLogHas(t, "binary "+bin+" already built — skipping build") + if strings.Contains(got.log, " $ ") { + t.Errorf("build ran a command despite the binary being there\nlog:\n%s", got.log) + } + }) + + t.Run("missing binary runs every command in order", func(t *testing.T) { + tmp := t.TempDir() + order := filepath.Join(tmp, "order.txt") + p := &plan.Plan{Bin: filepath.Join(tmp, "bin", "stellar-rpc-deadbeef"), Steps: []plan.Step{{ + ID: "build", Kind: plan.KindBuild, Argv: [][]string{ + {"/bin/sh", "-c", `echo checkout >> "$1"`, "sh", order}, + {"/bin/sh", "-c", `echo make >> "$1"`, "sh", order}, + }, + }}} + got := walk(t, p, Options{}) + got.assertStatuses(t, StatusOK) + b, err := os.ReadFile(order) + if err != nil { + t.Fatalf("read %s: %v", order, err) + } + if want := []string{"checkout", "make"}; !slices.Equal(strings.Fields(string(b)), want) { + t.Errorf("commands ran as %q, want %v", b, want) + } + }) + + t.Run("a failed command stops the rest of the step", func(t *testing.T) { + tmp := t.TempDir() + marker := filepath.Join(tmp, "second.txt") + p := &plan.Plan{Bin: filepath.Join(tmp, "bin", "stellar-rpc-deadbeef"), Steps: []plan.Step{{ + ID: "build", Kind: plan.KindBuild, Argv: [][]string{ + {"/bin/sh", "-c", "exit 1"}, + {"/bin/sh", "-c", `: > "$1"`, "sh", marker}, + }, + }}} + got := walk(t, p, Options{}) + got.assertStatuses(t, StatusFailed) + assertGone(t, marker) + }) +} + +// TestExecuteSkipsTheEpilogue pins the division of labour: the tarball and the +// publish are in the plan, but Execute leaves them to the run wiring, which +// makes the tarball only after the final provenance writes. +func TestExecuteSkipsTheEpilogue(t *testing.T) { + tmp := t.TempDir() + marker := filepath.Join(tmp, "tarball.txt") + tarball := plan.Step{ + ID: "tarball", Kind: plan.KindTarball, + Argv: [][]string{{"/bin/sh", "-c", `: > "$1"`, "sh", marker}}, + } + p := &plan.Plan{Steps: []plan.Step{ + shLeg("leg", filepath.Join(tmp, "res", "leg"), `: > "$1/driver.csv"`), + tarball, + {ID: "publish", Kind: plan.KindPublish, Argv: [][]string{}, PublishURI: "gs://bucket/runs", Needs: []string{"tarball"}}, + }} + + got := walk(t, p, Options{}) + if got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + // Only the leg has a result, and no tar ran. + got.assertStatuses(t, StatusOK) + assertGone(t, marker) + + // The wiring runs the same step itself, out of the same plan. + if err := RunStep(tarball, io.Discard); err != nil { + t.Fatalf("RunStep(tarball): %v", err) + } + assertExists(t, marker) +} + +func TestOnStepDoneFiresForExecutedSteps(t *testing.T) { + tmp := t.TempDir() + bin := filepath.Join(tmp, "bin", "stellar-rpc-deadbeef") + p := &plan.Plan{Bin: bin, Steps: []plan.Step{ + {ID: "build", Kind: plan.KindBuild, Argv: [][]string{{"/bin/sh", "-c", `mkdir -p "$(dirname "$1")" && : > "$1"`, "sh", bin}}}, + shLeg("fail", filepath.Join(tmp, "res", "fail"), `exit 1`), + shLeg("dep", filepath.Join(tmp, "res", "dep"), `: > "$1/driver.csv"`), + }} + p.Steps[2].Needs = []string{"fail"} + + var seen []string + var buf bytes.Buffer + if _, err := Execute(p, Options{ + Output: &buf, + OnStepDone: func(s plan.Step, r StepResult) { + seen = append(seen, s.ID+"="+string(r.Status)) + }, + }); err == nil { + t.Fatalf("Execute returned nil error after a failed leg\nlog:\n%s", buf.String()) + } + // The skipped step is the one exception: it never ran, so nothing is done. + if want := []string{"build=ok", "fail=failed"}; !slices.Equal(seen, want) { + t.Errorf("OnStepDone saw %v, want %v", seen, want) + } +} + +func TestExecuteLegWithMissingBinaryIsFailedWithASentinel(t *testing.T) { + tmp := t.TempDir() + out := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run1") + p := &plan.Plan{Steps: []plan.Step{{ + ID: "leg", Kind: plan.KindLeg, Timed: true, OutDir: out, + Argv: [][]string{{filepath.Join(tmp, "bin", "stellar-rpc-deadbeef"), "bench-ingest", "cold"}}, + }}} + got := walk(t, p, Options{}) + got.assertStatuses(t, StatusFailed) + // The binary never started, so it wrote no invocation.json — the sentinel + // is the only record that this leg was attempted, which is exactly why the + // executor creates the out dir itself. + s := readSentinel(t, out) + if s.ExitCode != -1 || s.Error == "" { + t.Errorf("sentinel = %+v, want exit_code -1 and an error", s) + } +} + +func TestExecuteAllOKPrintsNoSummary(t *testing.T) { + tmp := t.TempDir() + p := &plan.Plan{Steps: []plan.Step{shLeg("a", filepath.Join(tmp, "a"), `: > "$1/driver.csv"`)}} + got := walk(t, p, Options{}) + if got.err != nil { + t.Fatalf("Execute: %v", got.err) + } + if strings.Contains(got.log, "campaign summary") { + t.Errorf("an all-ok campaign printed a summary\nlog:\n%s", got.log) + } +} + +// --- lock ------------------------------------------------------------------- + +func TestAcquireLock(t *testing.T) { + benchRoot := filepath.Join(t.TempDir(), "bench") + release, err := AcquireLock(benchRoot) + if err != nil { + t.Fatalf("first AcquireLock: %v", err) + } + if _, err := AcquireLock(benchRoot); err == nil { + t.Fatal("second AcquireLock succeeded while the lock was held") + } else if want := "another campaign is already running on this BENCH_ROOT"; !strings.Contains(err.Error(), want) { + t.Errorf("error = %q, want it to contain %q", err, want) + } + + release() + release2, err := AcquireLock(benchRoot) + if err != nil { + t.Fatalf("AcquireLock after release: %v", err) + } + release2() + // The lock file outlives the lock: deleting it would race the next campaign. + assertExists(t, filepath.Join(benchRoot, lockName)) +} + +// --- logging ------------------------------------------------------------------ + +func TestNotef(t *testing.T) { + var buf bytes.Buffer + Notef(&buf, "build %s → %s", "abc1234", "/bench/bin/stellar-rpc-abc1234") + want := regexp.MustCompile(`^== \[\d{2}:\d{2}:\d{2}\] build abc1234 → /bench/bin/stellar-rpc-abc1234\n$`) + if !want.MatchString(buf.String()) { + t.Errorf("Notef wrote %q, want it to match %s", buf.String(), want) + } +} + +func TestOpenCampaignLogAppends(t *testing.T) { + dir := t.TempDir() + for _, session := range []string{"first\n", "second\n"} { + f, err := OpenCampaignLog(dir) + if err != nil { + t.Fatalf("OpenCampaignLog: %v", err) + } + if _, err := f.WriteString(session); err != nil { + t.Fatalf("write: %v", err) + } + f.Close() + } + b, err := os.ReadFile(filepath.Join(dir, campaignLogName)) + if err != nil { + t.Fatalf("read log: %v", err) + } + if string(b) != "first\nsecond\n" { + t.Errorf("campaign.log = %q, want both sessions in order", b) + } +} + +// TestLegSentinelJSONShape pins the wire format: the sentinel is read by resume +// and by anything inspecting a bundle, so its keys are a contract. +func TestLegSentinelJSONShape(t *testing.T) { + tmp := t.TempDir() + out := filepath.Join(tmp, "res", "ingest-cold-ds-c0-run1") + p := &plan.Plan{Steps: []plan.Step{shLeg("ingest-cold-ds-c0-run1", out, `exit 0`)}} + if got := walk(t, p, Options{}); got.err != nil { + t.Fatalf("Execute: %v\nlog:\n%s", got.err, got.log) + } + b, err := os.ReadFile(filepath.Join(out, legSentinelName)) + if err != nil { + t.Fatalf("read sentinel: %v", err) + } + var raw map[string]any + if err := json.Unmarshal(b, &raw); err != nil { + t.Fatalf("unmarshal sentinel: %v", err) + } + want := []string{"argv", "duration_ns", "exit_code", "finished_at", "id", "schema_version", "started_at"} + got := make([]string, 0, len(raw)) + for k := range raw { + got = append(got, k) + } + slices.Sort(got) + if !slices.Equal(got, want) { + t.Errorf("sentinel keys = %v, want %v", got, want) + } + // error is omitted on success, and only then. + if _, ok := raw["error"]; ok { + t.Errorf("a successful leg recorded an error: %s", b) + } +} diff --git a/runner/internal/run/source.go b/runner/internal/run/source.go new file mode 100644 index 0000000..88036b3 --- /dev/null +++ b/runner/internal/run/source.go @@ -0,0 +1,57 @@ +package run + +import ( + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" +) + +// EnsureSrc converges the persistent build clone at src onto repo: clone once, +// then per campaign point origin at repo (it may have changed since the clone +// was made), fetch its branches and tags, and hard-reset. Gitignored build +// caches (cargo target/, Go cache) survive the reset — clean -fd, deliberately +// no -x — so rebuilding a nearby commit is incremental. repo itself is never +// modified. +func EnsureSrc(src, repo string, out io.Writer) error { + if _, err := os.Stat(filepath.Join(src, ".git")); err != nil { + if err := runCommand([]string{"git", "clone", repo, src}, nil, out); err != nil { + return fmt.Errorf("clone %s into %s: %w", repo, src, err) + } + } + for _, argv := range [][]string{ + {"git", "-C", src, "remote", "set-url", "origin", repo}, + {"git", "-C", src, "fetch", "-q", "--prune", "origin", + "+refs/heads/*:refs/remotes/origin/*", "+refs/tags/*:refs/tags/*"}, + {"git", "-C", src, "reset", "-q", "--hard"}, + {"git", "-C", src, "clean", "-qfd"}, + } { + if err := runCommand(argv, nil, out); err != nil { + return fmt.Errorf("%s: %w", strings.Join(argv, " "), err) + } + } + return nil +} + +// ResolveRef returns the full commit ref resolves to inside src. +// Remote-tracking branches are tried first so a stale local ref never shadows +// the fetched branch tip; the fallback covers tags and raw hashes. +func ResolveRef(src, ref string) (string, error) { + // Without this guard git would search upwards from src and answer out of + // whatever repository happens to contain it. + if _, err := os.Stat(filepath.Join(src, ".git")); err != nil { + return "", fmt.Errorf("no build clone at %s", src) + } + for _, rev := range []string{"refs/remotes/origin/" + ref + "^{commit}", ref + "^{commit}"} { + out, err := exec.Command("git", "-C", src, "rev-parse", "--verify", "--quiet", rev).Output() + if err != nil { + continue + } + if sha := strings.TrimSpace(string(out)); len(sha) >= 8 { + return sha, nil + } + } + return "", fmt.Errorf("ref '%s' does not resolve to a commit in %s", ref, src) +} diff --git a/runner/internal/run/source_test.go b/runner/internal/run/source_test.go new file mode 100644 index 0000000..6b5e174 --- /dev/null +++ b/runner/internal/run/source_test.go @@ -0,0 +1,211 @@ +package run + +import ( + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" +) + +// git runs a git command in dir and fails the test if it does not succeed. +// Identity and hooks come from the flags, not the machine: the test must behave +// the same on a devbox with a global gitconfig and on a bare CI runner. +func git(t *testing.T, dir string, args ...string) string { + t.Helper() + argv := append([]string{ + "-c", "user.name=bench", "-c", "user.email=bench@example.com", + "-c", "commit.gpgsign=false", "-c", "init.defaultBranch=main", + }, args...) + cmd := exec.Command("git", argv...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_SYSTEM=/dev/null") + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %s in %s: %v\n%s", strings.Join(args, " "), dir, err, out) + } + return strings.TrimSpace(string(out)) +} + +// commit writes a file and commits it, returning the new commit's sha. +func commit(t *testing.T, repo, name, body string) string { + t.Helper() + mustWrite(t, filepath.Join(repo, name), body) + git(t, repo, "add", "-A") + git(t, repo, "commit", "-q", "-m", "add "+name) + return git(t, repo, "rev-parse", "HEAD") +} + +// originRepo is a local stellar-rpc stand-in: one commit, one gitignore, on +// branch main. +func originRepo(t *testing.T) (dir, head string) { + t.Helper() + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not installed") + } + dir = filepath.Join(t.TempDir(), "origin") + mustMkdir(t, dir) + git(t, dir, "init", "-q") + mustWrite(t, filepath.Join(dir, ".gitignore"), "target/\n") + return dir, commit(t, dir, "README", "first\n") +} + +func TestEnsureSrcClonesThenFetches(t *testing.T) { + origin, first := originRepo(t) + src := filepath.Join(t.TempDir(), "src") + + if err := EnsureSrc(src, origin, io.Discard); err != nil { + t.Fatalf("first EnsureSrc: %v", err) + } + assertExists(t, filepath.Join(src, ".git")) + got, err := ResolveRef(src, "main") + if err != nil { + t.Fatalf("ResolveRef after clone: %v", err) + } + if got != first { + t.Errorf("main = %s, want the origin head %s", got, first) + } + + // The origin moves on, as feature/full-history does between campaigns. + second := commit(t, origin, "NOTES", "second\n") + if err := EnsureSrc(src, origin, io.Discard); err != nil { + t.Fatalf("second EnsureSrc: %v", err) + } + got, err = ResolveRef(src, "main") + if err != nil { + t.Fatalf("ResolveRef after fetch: %v", err) + } + // This is the stale-local-ref case: the clone's own main still points at + // the first commit (a fetch updates only the remote-tracking refs), and + // resolving must follow origin/main, not it. + if local := git(t, src, "rev-parse", "refs/heads/main"); local != first { + t.Fatalf("test premise broken: local main = %s, want the stale %s", local, first) + } + if got != second { + t.Errorf("main = %s, want the fetched tip %s (a stale local ref shadowed it)", got, second) + } +} + +func TestEnsureSrcKeepsBuildCaches(t *testing.T) { + origin, _ := originRepo(t) + src := filepath.Join(t.TempDir(), "src") + if err := EnsureSrc(src, origin, io.Discard); err != nil { + t.Fatalf("EnsureSrc: %v", err) + } + + // target/ is gitignored (a cargo build cache); scratch.go is merely + // untracked. The reset must keep the first and drop the second, which is + // what clean -fd without -x buys: rebuilding a nearby commit stays + // incremental. + mustWrite(t, filepath.Join(src, "target", "libpreflight.a"), "cached") + mustWrite(t, filepath.Join(src, "scratch.go"), "package main") + mustWrite(t, filepath.Join(src, "README"), "locally edited\n") + + if err := EnsureSrc(src, origin, io.Discard); err != nil { + t.Fatalf("second EnsureSrc: %v", err) + } + assertExists(t, filepath.Join(src, "target", "libpreflight.a")) + assertGone(t, filepath.Join(src, "scratch.go")) + if b, err := os.ReadFile(filepath.Join(src, "README")); err != nil || string(b) != "first\n" { + t.Errorf("README = %q (err %v), want the reset content", b, err) + } +} + +func TestEnsureSrcRepointsOrigin(t *testing.T) { + first, _ := originRepo(t) + src := filepath.Join(t.TempDir(), "src") + if err := EnsureSrc(src, first, io.Discard); err != nil { + t.Fatalf("EnsureSrc: %v", err) + } + + // A campaign later points repo at a different checkout — a fork, or the + // operator's own work in progress. The clone follows it. + second, secondHead := originRepo(t) + git(t, second, "checkout", "-q", "-b", "feature/full-history") + secondHead = commit(t, second, "FEATURE", "wip\n") + if err := EnsureSrc(src, second, io.Discard); err != nil { + t.Fatalf("EnsureSrc onto the second origin: %v", err) + } + got, err := ResolveRef(src, "feature/full-history") + if err != nil { + t.Fatalf("ResolveRef: %v", err) + } + if got != secondHead { + t.Errorf("feature/full-history = %s, want %s from the new origin", got, secondHead) + } +} + +func TestEnsureSrcPrintsItsCommands(t *testing.T) { + origin, _ := originRepo(t) + src := filepath.Join(t.TempDir(), "src") + var buf strings.Builder + if err := EnsureSrc(src, origin, &buf); err != nil { + t.Fatalf("EnsureSrc: %v", err) + } + for _, want := range []string{ + " $ git clone " + origin + " " + src, + " $ git -C " + src + " remote set-url origin " + origin, + " $ git -C " + src + " fetch -q --prune origin +refs/heads/*:refs/remotes/origin/* +refs/tags/*:refs/tags/*", + " $ git -C " + src + " reset -q --hard", + " $ git -C " + src + " clean -qfd", + } { + if !strings.Contains(buf.String(), want) { + t.Errorf("log missing %q, got:\n%s", want, buf.String()) + } + } +} + +func TestEnsureSrcFailsOnAnUnreachableRepo(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not installed") + } + tmp := t.TempDir() + err := EnsureSrc(filepath.Join(tmp, "src"), filepath.Join(tmp, "no-such-repo"), io.Discard) + if err == nil { + t.Fatal("EnsureSrc succeeded on a repo that does not exist") + } + if !strings.Contains(err.Error(), "clone") { + t.Errorf("error = %v, want it to name the clone", err) + } +} + +func TestResolveRef(t *testing.T) { + origin, head := originRepo(t) + git(t, origin, "tag", "v1.2.3") + src := filepath.Join(t.TempDir(), "src") + if err := EnsureSrc(src, origin, io.Discard); err != nil { + t.Fatalf("EnsureSrc: %v", err) + } + + for _, tc := range []struct{ name, ref string }{ + {"branch", "main"}, + {"tag", "v1.2.3"}, + {"full sha", head}, + {"short sha", head[:8]}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := ResolveRef(src, tc.ref) + if err != nil { + t.Fatalf("ResolveRef(%s): %v", tc.ref, err) + } + if got != head { + t.Errorf("ResolveRef(%s) = %s, want %s", tc.ref, got, head) + } + }) + } + + t.Run("unknown ref", func(t *testing.T) { + if got, err := ResolveRef(src, "no/such/branch"); err == nil { + t.Errorf("ResolveRef = %s, want an error", got) + } + }) + + t.Run("no clone at all", func(t *testing.T) { + // Not merely "git fails here": without the .git guard, git would search + // upwards and answer out of whatever repository contains the path. + if got, err := ResolveRef(filepath.Join(src, "cmd"), "main"); err == nil { + t.Errorf("ResolveRef = %s, want an error naming the missing clone", got) + } + }) +} From ce68c9c6ab8fbb34266d5a57e5dec95795bb12c4 Mon Sep 17 00:00:00 2001 From: Marwen Abid Date: Sat, 1 Aug 2026 15:35:46 -0700 Subject: [PATCH 2/2] review: lock and clone errors name their real cause; log doc drops the bash file --- runner/internal/run/lock.go | 6 +++++- runner/internal/run/log.go | 2 +- runner/internal/run/source.go | 3 +++ 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/runner/internal/run/lock.go b/runner/internal/run/lock.go index db788ae..b71920b 100644 --- a/runner/internal/run/lock.go +++ b/runner/internal/run/lock.go @@ -1,6 +1,7 @@ package run import ( + "errors" "fmt" "os" "path/filepath" @@ -28,7 +29,10 @@ func AcquireLock(benchRoot string) (release func(), err error) { } if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { f.Close() - return nil, fmt.Errorf("another campaign is already running on this BENCH_ROOT (held lock: %s)", path) + if errors.Is(err, syscall.EWOULDBLOCK) { + return nil, fmt.Errorf("another campaign is already running on this BENCH_ROOT (held lock: %s)", path) + } + return nil, fmt.Errorf("lock %s: %w", path, err) } // The lock lives on the open file description, so the fd stays open until // release; closing it anywhere earlier would drop the lock silently. diff --git a/runner/internal/run/log.go b/runner/internal/run/log.go index 618a6e8..c253206 100644 --- a/runner/internal/run/log.go +++ b/runner/internal/run/log.go @@ -13,7 +13,7 @@ const campaignLogName = "campaign.log" // Notef prints a bash-style note — `== [HH:MM:SS] msg`, the clock in UTC — // the line format every operator reading a campaign log already knows from -// campaign.sh's note(). +// the note() of the bash campaign runner this package replaces. func Notef(w io.Writer, format string, args ...any) { fmt.Fprintf(w, "== [%s] %s\n", time.Now().UTC().Format("15:04:05"), fmt.Sprintf(format, args...)) } diff --git a/runner/internal/run/source.go b/runner/internal/run/source.go index 88036b3..47230cc 100644 --- a/runner/internal/run/source.go +++ b/runner/internal/run/source.go @@ -17,6 +17,9 @@ import ( // modified. func EnsureSrc(src, repo string, out io.Writer) error { if _, err := os.Stat(filepath.Join(src, ".git")); err != nil { + if !os.IsNotExist(err) { + return fmt.Errorf("inspect build clone at %s: %w", src, err) + } if err := runCommand([]string{"git", "clone", repo, src}, nil, out); err != nil { return fmt.Errorf("clone %s into %s: %w", repo, src, err) }