Add RPC database backfill performance evaluation leg

.github/workflows/ec2-leg.yml

@@ -210,8 +210,8 @@ jobs:
           echo "::error::SSM agent never registered for $INSTANCE_ID — verify AmazonSSMManagedInstanceCore is attached to the stellar-rpc-ci-load-test role"
           exit 1
 
-      # Polls S3 for the result object and relays verdict/passed + the result
-      # location (bucket/key) as step outputs the coordinator reads.
+      # Polls S3 for the result object and relays found/passed as step outputs
+      # for the fail gate below.
       # RUN_ID must match the box preamble's: gather skips result objects from
       # other attempts (a re-run shares RESULT_KEY with its predecessors).
       - name: Poll for results

.github/workflows/load-test-coordinator.yml

@@ -19,6 +19,11 @@ on:
         required: false
         type: string
         default: ''
+      pr_number:
+        description: 'PR to report results to (default: inferred from the tested branch)'
+        required: false
+        type: string
+        default: ''
 
 # Keyed on the release ref so a new push to the same branch supersedes an
 # in-flight test. Each leg's self-terminate ceiling backstops its box.
@@ -50,14 +55,21 @@ jobs:
         env:
           GH_TOKEN: ${{ github.token }}
           REF: ${{ inputs.target_ref || github.ref_name }}
+          PR: ${{ inputs.pr_number }}
         run: |
           SHA="$(git rev-parse HEAD)"
-          PR="$(gh pr list --repo "${{ github.repository }}" --state open \
-                  --base main --head "$REF" --json number --jq '.[0].number // ""' 2>/dev/null || true)"
+          if [ -z "$PR" ]; then
+            PR="$(gh pr list --repo "${{ github.repository }}" --state open \
+                    --head "$REF" --json number --jq '.[0].number // ""' 2>/dev/null || true)"
+          fi
+          LEGS='[
+            {"label":"Apply-load ingestion","run_label":"apply-load","script":"cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/run-load-test.sh","budget":225},
+            {"label":"Backfill ingestion","run_label":"backfill","script":"cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh","budget":345}
+          ]'
           {
             echo "target_sha=$SHA"
             echo "pr_number=$PR"
-            echo 'legs=[{"label":"Apply-load ingestion","run_label":"apply-load","script":"cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/run-load-test.sh"}]'
+            echo "legs=$(jq -c . <<<"$LEGS")"
           } >> "$GITHUB_OUTPUT"
 
   # The fan-out: one matrix entry per leg.
@@ -74,6 +86,7 @@ jobs:
       target_ref: ${{ needs.plan.outputs.target_sha }}
       run_label: ${{ matrix.leg.run_label }}
       leg_script: ${{ matrix.leg.script }}
+      budget_minutes: ${{ matrix.leg.budget }}
 
   report:
     name: Aggregate + report

cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh

@@ -0,0 +1,9 @@
+# Backfill ingestion leg. Concatenated after bootstrap-common.sh in the rendered
+# EC2 user-data, so it relies on that file's env, helpers, bootstrap_box, and
+# run_leg. It hands off to `runner instantiate`, which builds stellar-rpc and
+# times a backfill run against the pubnet datastore. The other half, `runner 
+# gather`, polls S3 for the result object.
+LEG_TITLE="Backfill ingestion"
+
+bootstrap_box
+run_leg ./cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner

cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/instantiate.go

@@ -0,0 +1,223 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+
+	"github.com/stellar/go-stellar-sdk/ingest/ledgerbackend"
+
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness"
+)
+
+const (
+	// runner runs w/ cwd = repo root, so paths are relative to there
+	legDir   = "cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test"
+	corePath = "/usr/local/bin/stellar-core" // fetched from S3
+)
+
+const ledgerThreshold = 384 // mirrors ingest.ledgerThreshold in backfill.go
+
+// backfillDoneRe matches the terminal line emitted on backfill's completion
+var backfillDoneRe = regexp.MustCompile(`Backfill process complete, ledgers \[(\d+) -> (\d+)\]`)
+
+// instantiate is the instance's backfill task: it fetches + builds test fixtures,
+// runs a timed backfill, then publishes the verdict.
+func instantiate(ctx context.Context) error {
+	var (
+		bucket      = harness.Env("BUCKET", "stellar-rpc-ci-load-test")
+		region      = harness.Env("REGION", "us-east-1")
+		workDir     = harness.Env("WORK_DIR", "/data")
+		resultsFile = harness.Env("RESULTS_FILE", "/tmp/results.md")
+		resultKey   = os.Getenv("RESULT_KEY")
+		targetSHA   = os.Getenv("TARGET_SHA")
+		runID       = harness.Env("RUN_ID", "manual")
+		// ~1 day by default for cheap test runs; the full week is 120960.
+		retention = harness.Env("HISTORY_RETENTION_WINDOW", "17280")
+		deadline  = harness.Env("BACKFILL_DEADLINE", "4h")
+
+		binaryPath = filepath.Join(workDir, "stellar-rpc-bin") // built here (the repo checkout is in WORK_DIR)
+	)
+	repoRoot, err := os.Getwd()
+	if err != nil {
+		return err
+	}
+	bail := func(format string, args ...any) error {
+		return harness.BailInstance(resultsFile, "Backfill ingestion", runID, targetSHA, fmt.Sprintf(format, args...))
+	}
+
+	want, err := strconv.Atoi(retention) // used to compare against ingested ledgers below
+	if err != nil {
+		return bail("parsing retention window %q: %v", retention, err)
+	}
+
+	awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
+	if err != nil {
+		return bail("loading AWS config: %v", err)
+	}
+	fetch := &harness.S3Fetcher{Client: s3.NewFromConfig(awsCfg), Bucket: bucket}
+
+	if err := fetch.FetchVerified(ctx, "core/stellar-core.zst", corePath, true, "stellar-core"); err != nil {
+		return bail("%v", err)
+	}
+	if err := os.Chmod(corePath, 0o755); err != nil {
+		return bail("chmod stellar-core: %v", err)
+	}
+
+	logger.Infof("building stellar-rpc")
+	if err := harness.RunStreaming(ctx, repoRoot, nil, 40, "make", "build-libs"); err != nil {
+		return bail("make build-libs failed: %v", err)
+	}
+	if err := harness.RunStreaming(ctx, repoRoot, nil, 40,
+		"go", "build", "-o", binaryPath, "./cmd/stellar-rpc"); err != nil {
+		return bail("go build failed: %v", err)
+	}
+
+	// fetch + write core config from SDK
+	coreCfg := filepath.Join(workDir, "captive-core-pubnet.cfg")
+	if err := os.WriteFile(coreCfg, ledgerbackend.PubnetDefaultConfig, 0o644); err != nil {
+		return bail("writing captive-core config: %v", err)
+	}
+
+	cfgPath, err := renderConfig(repoRoot, workDir, coreCfg, retention)
+	if err != nil {
+		return bail("rendering config: %v", err)
+	}
+
+	dl, err := time.ParseDuration(deadline)
+	if err != nil {
+		return bail("invalid BACKFILL_DEADLINE %q: %v", deadline, err)
+	}
+	logger.Infof("starting backfill (retention=%s, deadline=%s)", retention, deadline)
+	elapsed, lo, hi, err := runBackfill(ctx, dl, binaryPath, cfgPath)
+	if err != nil {
+		return bail("%v", err)
+	}
+	ingested := hi - lo + 1
+	if ingested+ledgerThreshold < want {
+		return bail("backfill reported complete but ingested %d of %s ledgers", ingested, retention)
+	}
+	logger.Infof("backfill complete: %d ledgers [%d -> %d] in %s", ingested, lo, hi, elapsed.Round(time.Second))
+
+	md := renderMarkdown(targetSHA, retention, lo, hi, ingested, elapsed)
+	if err := os.WriteFile(resultsFile, []byte(md), 0o644); err != nil {
+		return bail("writing results: %v", err)
+	}
+	if err := harness.PublishResult(
+		ctx, fetch.Client, bucket, resultKey, "ok", runID, targetSHA, resultsFile, ""); err != nil {
+		return bail("publishing result: %v", err)
+	}
+	return nil
+}
+
+// renderConfig fills the config template's ${...} placeholders (box paths + the
+// retention window) via os.Expand
+func renderConfig(repoRoot, workDir, coreCfg, retention string) (string, error) {
+	tmpl, err := os.ReadFile(filepath.Join(repoRoot, legDir, "testdata", "backfill-pubnet.toml.tmpl"))
+	if err != nil {
+		return "", err
+	}
+	mapping := func(in string) string {
+		switch in {
+		case "CAPTIVE_CORE_CONFIG_PATH":
+			return coreCfg
+		case "CAPTIVE_CORE_STORAGE_PATH":
+			return filepath.Join(workDir, "core-storage")
+		case "DB_PATH":
+			return filepath.Join(workDir, "backfill.sqlite")
+		case "STELLAR_CORE_BINARY_PATH":
+			return corePath
+		case "HISTORY_RETENTION_WINDOW":
+			return retention
+		default:
+			return "${" + in + "}" // leave unknown placeholders intact
+		}
+	}
+	body := os.Expand(string(tmpl), mapping)
+	cfgPath := filepath.Join(workDir, "backfill-rpc.toml")
+	if err := os.WriteFile(cfgPath, []byte(body), 0o644); err != nil {
+		return "", err
+	}
+	return cfgPath, nil
+}
+
+// runBackfill launches the daemon and streams its output (teeing to the box log)
+// until the backfill-complete line fires, recording the wall-clock
+func runBackfill(ctx context.Context, deadline time.Duration, binary, cfgPath string) (time.Duration, int, int, error) {
+	runCtx, cancel := context.WithTimeout(ctx, deadline)
+	defer cancel()
+
+	cmd := exec.CommandContext(runCtx, binary, "--config-path", cfgPath)
+	// hide this box's IMDS creds as the public datalake 403s signed requests
+	cmd.Env = append(os.Environ(), "AWS_EC2_METADATA_DISABLED=true")
+	pr, pw, err := os.Pipe()
+	if err != nil {
+		return 0, 0, 0, err
+	}
+	cmd.Stdout, cmd.Stderr = pw, pw
+
+	start := time.Now()
+	if err := cmd.Start(); err != nil {
+		pw.Close()
+		pr.Close()
+		return 0, 0, 0, fmt.Errorf("starting daemon: %w", err)
+	}
+	pw.Close() // the child holds the write end and we read until it dies
+	defer pr.Close()
+
+	var elapsed time.Duration
+	var lo, hi int
+	scanner := bufio.NewScanner(pr)
+	scanner.Buffer(make([]byte, 64*1024), 1024*1024)
+	for scanner.Scan() {
+		line := scanner.Text()
+		fmt.Fprintln(os.Stderr, line) // tee to the box user-data log (SSM debug tail)
+		if m := backfillDoneRe.FindStringSubmatch(line); m != nil {
+			elapsed = time.Since(start)
+			lo, _ = strconv.Atoi(m[1])
+			hi, _ = strconv.Atoi(m[2])
+			cancel() // stop the daemon before it starts live ingestion
+			break
+		}
+	}
+	scanErr := scanner.Err()
+	if scanErr != nil {
+		cancel()
+	}
+	_ = cmd.Wait() // reap; a kill from cancel surfaces here and is expected
+
+	if elapsed == 0 {
+		if scanErr != nil {
+			return 0, 0, 0, fmt.Errorf("reading daemon output: %w", scanErr)
+		}
+		return 0, 0, 0, fmt.Errorf("daemon exited or hit the %s deadline before backfill completed", deadline)
+	}
+	return elapsed, lo, hi, nil
+}
+
+func renderMarkdown(sha, retention string, lo, hi, ingested int, elapsed time.Duration) string {
+	shortSHA := sha
+	if len(shortSHA) > 12 {
+		shortSHA = shortSHA[:12]
+	}
+	lps := 0.0
+	if s := elapsed.Seconds(); s > 0 {
+		lps = float64(ingested) / s
+	}
+	return fmt.Sprintf("### ⏳ Backfill ingestion — `%s`\n\n"+
+		"| Metric | Value |\n|---|---|\n"+
+		"| Ledgers ingested | %d (`[%d -> %d]`) |\n"+
+		"| Retention window | %s |\n"+
+		"| Wall-clock | %s |\n"+
+		"| Ledgers/sec | %.1f |\n",
+		shortSHA, ingested, lo, hi, retention, elapsed.Round(time.Second), lps)
+}

cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/main.go

@@ -0,0 +1,8 @@
+// Command runner runs the backfill ingestion benchmark on the box.
+package main
+
+import "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness"
+
+var logger = harness.NewLogger()
+
+func main() { harness.Run(instantiate) }

cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/testdata/backfill-pubnet.toml.tmpl

@@ -0,0 +1,36 @@
+# To fill in and use by the backfill leg's runner (instantiate.go renderConfig),
+# which substitutes the placeholders below via os.Expand before the daemon reads
+# it -- not a standalone config.
+
+NETWORK_PASSPHRASE = "Public Global Stellar Network ; September 2015"
+HISTORY_ARCHIVE_URLS = ["https://history.stellar.org/prd/core-live/core_live_001"]
+ENDPOINT = "localhost:8000"
+LOG_LEVEL = "info"
+
+# Captive core is constructed at startup (so the binary + config must exist)
+# even though backfill itself reads from the datastore below.
+CAPTIVE_CORE_CONFIG_PATH = "${CAPTIVE_CORE_CONFIG_PATH}"
+STELLAR_CORE_BINARY_PATH = "${STELLAR_CORE_BINARY_PATH}"
+CAPTIVE_CORE_STORAGE_PATH = "${CAPTIVE_CORE_STORAGE_PATH}"
+CHECKPOINT_FREQUENCY = 64
+
+DB_PATH = "${DB_PATH}"
+
+# Backfill the retention window from the datastore on startup, then stop (the
+# leg kills the daemon at the completion log line).
+BACKFILL = true
+SERVE_LEDGERS_FROM_DATASTORE = true
+HISTORY_RETENTION_WINDOW = ${HISTORY_RETENTION_WINDOW}
+INGESTION_TIMEOUT = "50m0s"
+
+[datastore_config]
+  type = "S3"
+
+  [datastore_config.params]
+    destination_bucket_path = "aws-public-blockchain/v1.1/stellar/ledgers/pubnet"
+    region = "us-east-2"
+
+  [datastore_config.schema]
+    FileExtension = "zst"
+    files_per_partition = 64000
+    ledgers_per_file = 1