From 69ce6efefbc8e759c0e3649b9cfda55d2cf4144f Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Mon, 31 Aug 2026 16:29:49 -0400 Subject: [PATCH 1/3] feat(membench): add push memory benchmark harness objgitd already served net/http/pprof and the Go runtime collectors on the metrics listener, but nothing drove a repeatable push workload against them, so per-push memory cost was guesswork. membench owns a daemon, pushes one mirror-cloned repository into many fresh UUID-named repos, and samples /proc plus /metrics every 250ms, capturing a heap profile whenever resident memory sets a new high-water mark. It runs an idle baseline, a sequential phase, and a concurrency sweep, so per-push cost, retention between pushes, and the slope against concurrency are separable. Two measurement details are load-bearing. Peaks come from VmHWM rather than sampled VmRSS, because the kernel's own high-water mark cannot be missed between ticks. Peak captures use gc=0, since forcing a collection would move the number that triggered the capture. A failed /metrics scrape keeps its /proc numbers but records no runtime figures: the last tick before shutdown routinely fails, and treating its zero values as a measurement reported a settled heap of zero. Repositories are never deleted. The run writes repos.txt and leaves cleanup to the operator. Assisted-by: Claude Opus 5 via Claude Code Claude-Session: https://claude.ai/code/session_01PmugaKamcmXCxM4u4w29iD Signed-off-by: Xe Iaso --- AGENTS.md | 4 +- cmd/membench/main.go | 513 +++++++++++++++++++++++++++++++++ cmd/membench/metrics.go | 80 +++++ cmd/membench/metrics_test.go | 77 +++++ cmd/membench/proc.go | 83 ++++++ cmd/membench/proc_test.go | 77 +++++ cmd/membench/report.go | 374 ++++++++++++++++++++++++ cmd/membench/report_test.go | 313 ++++++++++++++++++++ cmd/membench/sampler.go | 224 ++++++++++++++ cmd/membench/uuid.go | 21 ++ docs/usage/memory-benchmark.md | 124 ++++++++ 11 files changed, 1889 insertions(+), 1 deletion(-) create mode 100644 cmd/membench/main.go create mode 100644 cmd/membench/metrics.go create mode 100644 cmd/membench/metrics_test.go create mode 100644 cmd/membench/proc.go create mode 100644 cmd/membench/proc_test.go create mode 100644 cmd/membench/report.go create mode 100644 cmd/membench/report_test.go create mode 100644 cmd/membench/sampler.go create mode 100644 cmd/membench/uuid.go create mode 100644 docs/usage/memory-benchmark.md diff --git a/AGENTS.md b/AGENTS.md index d44dac6..1a9757b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,6 +66,7 @@ Notes on the tests and on configuration: | `internal/mountfs`, `internal/treefs`, `internal/kefkash` | The hook sandbox filesystem and shell wiring. | | `internal/metrics` | Every Prometheus vector, plus thin helpers. | | `internal/slog.go` | JSON handler init. | +| `cmd/membench/` | Push memory benchmark harness. Not shipped; see `docs/usage/memory-benchmark.md`. | ## Architecture @@ -85,7 +86,8 @@ Two more directories carry detail: - `docs/reference/` — the `.cue` binary layout, the failure modes, and the build order. -- `docs/usage/` — how to use a feature, such as a hook script. +- `docs/usage/` — how to use a feature, such as a hook script, or the push + memory benchmark. ## Conventions diff --git a/cmd/membench/main.go b/cmd/membench/main.go new file mode 100644 index 0000000..316edea --- /dev/null +++ b/cmd/membench/main.go @@ -0,0 +1,513 @@ +// Command membench measures how much memory objgitd uses while it takes +// pushes. It starts a daemon it owns, pushes one real repository into many +// fresh ones, and samples the daemon's resident set and Go heap the whole time, +// grabbing a pprof heap profile every time memory sets a new high-water mark. +// +// The run has three phases. A baseline establishes what the daemon costs +// sitting idle. A sequential phase pushes one repository at a time with idle +// gaps, which isolates the per-push cost and shows whether memory comes back +// down afterwards. A concurrency sweep then runs K pushes at once for a few +// values of K, which gives the slope you actually size a machine with. +// +// Every repository is created fresh under a UUID, so no push is ever measured +// against a repository that already holds its objects. The harness never +// deletes them; it writes repos.txt and leaves cleanup to you. +package main + +import ( + "context" + "flag" + "fmt" + "log/slog" + "net/http" + "os" + "os/exec" + "os/signal" + "path/filepath" + "runtime" + "strconv" + "strings" + "syscall" + "time" + + "github.com/facebookgo/flagenv" + "github.com/tigrisdata/objgit/internal" + "golang.org/x/sync/errgroup" +) + +// Phase names, as they appear in the phase column of samples.csv. +const ( + phaseBaseline = "baseline" + phaseIdle = "idle" + phaseSeq = "seq" + phaseConc = "conc" +) + +var ( + repoPath = flag.String("repo", filepath.Join(os.Getenv("HOME"), "Code/Xe/x"), "repository to push; it is mirror-cloned once and never written to") + outBase = flag.String("out", "", "parent directory for run output; empty uses the OS temp directory") + org = flag.String("org", "benchtest", "org segment every benchmark repository is created under") + + seqPushes = flag.Int("seq-pushes", 5, "number of sequential pushes, each to a fresh repository") + concSteps = flag.String("conc-steps", "1,2,4,8", "comma-separated concurrency levels to sweep; empty skips the sweep") + + sampleEvery = flag.Duration("sample-interval", 250*time.Millisecond, "how often to sample /proc and /metrics") + idleGap = flag.Duration("idle-gap", 5*time.Second, "idle time between pushes, so memory has a chance to settle") + baselineFor = flag.Duration("baseline", 10*time.Second, "how long to sample the idle daemon before pushing anything") + windowSlack = flag.Duration("window-slack", 0, "widen each push's measurement window by this much on both sides, so work finishing after git exits still counts against it; zero uses -idle-gap") + peakGrowth = flag.Float64("peak-growth", 0.05, "fractional rise in resident set that triggers a heap profile capture") + peakCooldown = flag.Duration("peak-cooldown", 2*time.Second, "minimum time between two peak captures") + + daemonBinary = flag.String("daemon-binary", "", "prebuilt objgitd to run; empty builds ./cmd/objgitd") + daemonHTTPBind = flag.String("daemon-http-bind", "127.0.0.1:8080", "address the daemon under test serves smart HTTP on") + daemonMetrics = flag.String("daemon-metrics-bind", "127.0.0.1:9090", "address the daemon under test serves /metrics and /debug/pprof on") + daemonAllowHooks = flag.Bool("daemon-allow-hooks", false, "run push hooks in the daemon under test; off by default so hook cost is not mistaken for push cost") + daemonReadyWait = flag.Duration("daemon-ready-wait", 60*time.Second, "how long to wait for the daemon to answer /metrics before giving up") + + slogLevel = flag.String("slog-level", "INFO", "log level (DEBUG, INFO, WARN, ERROR)") +) + +func main() { + flagenv.Parse() + flag.Parse() + + logger, err := internal.InitSlog(*slogLevel) + if err != nil { + fmt.Fprintf(os.Stderr, "bad -slog-level: %v\n", err) + os.Exit(1) + } + slog.SetDefault(logger) + + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + + if err := run(ctx); err != nil { + slog.Error("benchmark failed", "err", err) + os.Exit(1) + } +} + +func run(ctx context.Context) error { + steps, err := parseSteps(*concSteps) + if err != nil { + return err + } + + root, err := moduleRoot() + if err != nil { + return err + } + + startedAt := time.Now() + runDir := filepath.Join(outDir(), startedAt.Format("20060102-150405")) + if err := os.MkdirAll(runDir, 0o755); err != nil { + return fmt.Errorf("membench: can't create %s: %w", runDir, err) + } + slog.Info("run directory", "path", runDir) + + mirror := filepath.Join(runDir, "source.git") + if err := mirrorClone(ctx, *repoPath, mirror); err != nil { + return err + } + + packed, err := packBytes(mirror) + if err != nil { + return err + } + slog.Info("mirrored source repository", "path", mirror, "pack_bytes", packed) + + bin := *daemonBinary + if bin == "" { + bin = filepath.Join(runDir, "objgitd") + if err := buildDaemon(ctx, root, bin); err != nil { + return err + } + } + + // -pack-cache-dir names the parent the daemon makes its own cache directory + // under, so it has to exist before the daemon starts. + packCache := filepath.Join(runDir, "packcache") + if err := os.MkdirAll(packCache, 0o755); err != nil { + return fmt.Errorf("membench: can't create %s: %w", packCache, err) + } + + args := []string{ + "-http-bind", *daemonHTTPBind, + "-metrics-bind", *daemonMetrics, + "-ssh-bind=", + "-allow-push", + fmt.Sprintf("-allow-hooks=%t", *daemonAllowHooks), + "-pack-cache-dir", packCache, + } + + daemon, err := startDaemon(root, bin, filepath.Join(runDir, "daemon.log"), args) + if err != nil { + return err + } + defer stopDaemon(daemon) + + if err := waitReady(ctx, *daemonMetrics, *daemonReadyWait); err != nil { + return err + } + slog.Info("daemon ready", "pid", daemon.Process.Pid, "http", *daemonHTTPBind, "metrics", *daemonMetrics) + + s := newSampler(daemon.Process.Pid, *daemonMetrics, runDir, *sampleEvery, *peakCooldown, *peakGrowth) + sampleCtx, stopSampling := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + s.run(sampleCtx) + }() + + pushes, runErr := drive(ctx, s, mirror, steps) + + stopSampling() + <-done + + samples, profiles := s.snapshot() + meta := runMeta{ + StartedAt: startedAt, + FinishedAt: time.Now(), + Host: hostname(), + GoVersion: runtime.Version(), + GOGC: os.Getenv("GOGC"), + GOMEMLIMIT: os.Getenv("GOMEMLIMIT"), + NumCPU: runtime.NumCPU(), + SourceRepo: *repoPath, + PackBytes: packed, + Org: *org, + DaemonArgs: args, + SampleEvery: *sampleEvery, + WindowSlack: slack(), + RunDir: runDir, + } + + if err := writeCSV(filepath.Join(runDir, "samples.csv"), samples); err != nil { + return err + } + if err := writeRepoList(filepath.Join(runDir, "repos.txt"), *org, pushes); err != nil { + return err + } + if err := writeReport(filepath.Join(runDir, "report.md"), meta, samples, profiles, pushes); err != nil { + return err + } + + slog.Info("wrote results", + "report", filepath.Join(runDir, "report.md"), + "samples", len(samples), + "profiles", len(profiles), + "repos", len(pushes), + ) + fmt.Printf("\n%s\n\nRepositories left in the bucket are listed in %s.\n", + filepath.Join(runDir, "report.md"), filepath.Join(runDir, "repos.txt")) + + return runErr +} + +// drive walks the three phases. It returns whatever it managed to record even +// when a push fails, because a partial run with a heap profile in it is still +// worth reading. +func drive(ctx context.Context, s *sampler, mirror string, steps []int) ([]pushResult, error) { + var pushes []pushResult + + s.label(phaseBaseline, "") + slog.Info("sampling idle baseline", "for", *baselineFor) + if err := idle(ctx, *baselineFor); err != nil { + return pushes, err + } + if _, err := s.captureProfile(ctx, "heap", true, "heap-baseline.pb.gz", "idle baseline before any push", lastRSS(s)); err != nil { + slog.Warn("can't capture baseline heap profile", "err", err) + } + + for i := range *seqPushes { + if err := ctx.Err(); err != nil { + return pushes, err + } + + name, err := newUUID() + if err != nil { + return pushes, err + } + + s.label(phaseSeq, name) + slog.Info("sequential push", "n", i+1, "of", *seqPushes, "repo", *org+"/"+name) + pushes = append(pushes, doPush(ctx, mirror, phaseSeq, 1, name)) + + s.label(phaseIdle, "") + if err := idle(ctx, *idleGap); err != nil { + return pushes, err + } + } + + for _, k := range steps { + if err := ctx.Err(); err != nil { + return pushes, err + } + + names := make([]string, k) + for i := range names { + name, err := newUUID() + if err != nil { + return pushes, err + } + names[i] = name + } + + s.label(phaseConc, fmt.Sprintf("k=%d", k)) + slog.Info("concurrent push step", "k", k) + + results := make([]pushResult, k) + g, gCtx := errgroup.WithContext(ctx) + for i, name := range names { + g.Go(func() error { + results[i] = doPush(gCtx, mirror, phaseConc, k, name) + return nil + }) + } + _ = g.Wait() + pushes = append(pushes, results...) + + s.label(phaseIdle, "") + if err := idle(ctx, *idleGap); err != nil { + return pushes, err + } + } + + s.label(phaseIdle, "") + if err := idle(ctx, *idleGap); err != nil { + return pushes, err + } + if _, err := s.captureProfile(ctx, "heap", true, "heap-final.pb.gz", "settled heap after every push", lastRSS(s)); err != nil { + slog.Warn("can't capture final heap profile", "err", err) + } + if _, err := s.captureProfile(ctx, "allocs", false, "allocs-final.pb.gz", "total allocation over the whole run", lastRSS(s)); err != nil { + slog.Warn("can't capture allocation profile", "err", err) + } + + return pushes, nil +} + +func doPush(ctx context.Context, mirror, phase string, concurrency int, name string) pushResult { + url := fmt.Sprintf("http://%s/%s/%s.git", *daemonHTTPBind, *org, name) + + start := time.Now() + err := push(ctx, mirror, url) + end := time.Now() + + if err != nil { + slog.Error("push failed", "repo", *org+"/"+name, "err", err) + } else { + slog.Info("push done", "repo", *org+"/"+name, "took", end.Sub(start).Round(time.Millisecond)) + } + + return pushResult{Phase: phase, Concurrency: concurrency, Repo: name, Start: start, End: end, Err: err} +} + +// push sends every branch and tag to a repository that does not exist yet. +// The refspecs are forced because the harness does not care about history on +// the receiving side; it only cares what the daemon does with the pack. +func push(ctx context.Context, mirror, url string) error { + cmd := exec.CommandContext(ctx, "git", "-C", mirror, "push", "--quiet", url, + "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*") + + out, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("git push to %s: %w: %s", url, err, strings.TrimSpace(string(out))) + } + + return nil +} + +// mirrorClone copies the source repository once, so the benchmark never writes +// to the repository it was pointed at and concurrent pushes all read from the +// same immutable copy. +func mirrorClone(ctx context.Context, src, dst string) error { + cmd := exec.CommandContext(ctx, "git", "clone", "--mirror", src, dst) + + out, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("membench: can't mirror %s: %w: %s", src, err, strings.TrimSpace(string(out))) + } + + return nil +} + +func buildDaemon(ctx context.Context, root, out string) error { + cmd := exec.CommandContext(ctx, "go", "build", "-o", out, "./cmd/objgitd") + cmd.Dir = root + + res, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("membench: can't build objgitd: %w: %s", err, strings.TrimSpace(string(res))) + } + + slog.Info("built daemon under test", "path", out) + return nil +} + +// startDaemon runs objgitd with its working directory set to the module root, +// because objgitd loads its bucket and credentials from the .env file there. +// The context is deliberately not attached: shutdown goes through stopDaemon so +// the daemon gets the same SIGINT it would in production. +func startDaemon(root, bin, logPath string, args []string) (*exec.Cmd, error) { + log, err := os.Create(logPath) + if err != nil { + return nil, fmt.Errorf("membench: can't create %s: %w", logPath, err) + } + + cmd := exec.Command(bin, args...) + cmd.Dir = root + cmd.Stdout = log + cmd.Stderr = log + + if err := cmd.Start(); err != nil { + log.Close() + return nil, fmt.Errorf("membench: can't start %s: %w", bin, err) + } + + return cmd, nil +} + +func stopDaemon(cmd *exec.Cmd) { + if cmd.Process == nil { + return + } + + if err := cmd.Process.Signal(os.Interrupt); err != nil { + slog.Warn("can't interrupt daemon", "err", err) + } + + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + + select { + case <-done: + case <-time.After(15 * time.Second): + slog.Warn("daemon did not exit on SIGINT, killing it") + _ = cmd.Process.Kill() + <-done + } +} + +func waitReady(ctx context.Context, metricsAddr string, limit time.Duration) error { + url := "http://" + metricsAddr + "/metrics" + client := &http.Client{Timeout: 2 * time.Second} + deadline := time.Now().Add(limit) + + for time.Now().Before(deadline) { + if err := ctx.Err(); err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return fmt.Errorf("membench: can't build readiness request: %w", err) + } + + resp, err := client.Do(req) + if err == nil { + resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return nil + } + } + + if err := idle(ctx, 250*time.Millisecond); err != nil { + return err + } + } + + return fmt.Errorf("membench: daemon did not answer %s within %s", url, limit) +} + +// idle sleeps, but gives up as soon as the run is cancelled. +func idle(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + +func lastRSS(s *sampler) uint64 { + samples, _ := s.snapshot() + if len(samples) == 0 { + return 0 + } + return samples[len(samples)-1].Proc.VmRSS +} + +// parseSteps turns "1,2,4,8" into the concurrency levels to sweep. +func parseSteps(raw string) ([]int, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, nil + } + + var out []int + for _, part := range strings.Split(raw, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + n, err := strconv.Atoi(part) + if err != nil { + return nil, fmt.Errorf("membench: %q in -conc-steps is not a number: %w", part, err) + } + if n < 1 { + return nil, fmt.Errorf("membench: -conc-steps must be positive, got %d", n) + } + out = append(out, n) + } + + return out, nil +} + +// slack is how far outside a push's wall clock the harness still attributes +// memory to it. The daemon keeps uploading after git has exited, so the default +// is the whole idle gap that follows. +func slack() time.Duration { + if *windowSlack > 0 { + return *windowSlack + } + return *idleGap +} + +func outDir() string { + if *outBase != "" { + return *outBase + } + return filepath.Join(os.TempDir(), "membench") +} + +// moduleRoot walks up from the working directory to the directory holding +// go.mod. That is where objgitd's .env lives, and where `go build` has to run. +func moduleRoot() (string, error) { + dir, err := os.Getwd() + if err != nil { + return "", fmt.Errorf("membench: can't read working directory: %w", err) + } + + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir, nil + } + parent := filepath.Dir(dir) + if parent == dir { + return "", fmt.Errorf("membench: no go.mod above the working directory; run this from the objgit checkout") + } + dir = parent + } +} + +func hostname() string { + h, err := os.Hostname() + if err != nil { + return "unknown" + } + return h +} diff --git a/cmd/membench/metrics.go b/cmd/membench/metrics.go new file mode 100644 index 0000000..dbbf986 --- /dev/null +++ b/cmd/membench/metrics.go @@ -0,0 +1,80 @@ +package main + +import ( + "bufio" + "fmt" + "io" + "strconv" + "strings" +) + +// goMetrics is the subset of objgitd's /metrics this harness records. The Go +// runtime numbers say what the heap is doing; ProcessRSS is the same figure +// /proc reports, kept as a cross-check that the sampler is watching the pid it +// thinks it is. +type goMetrics struct { + HeapInuse uint64 + HeapSys uint64 + NextGC uint64 + Goroutines uint64 + ProcessRSS uint64 +} + +// field returns where a metric name lands, or nil for a metric this harness +// does not record. +func (m *goMetrics) field(name string) *uint64 { + switch name { + case "go_memstats_heap_inuse_bytes": + return &m.HeapInuse + case "go_memstats_heap_sys_bytes": + return &m.HeapSys + case "go_memstats_next_gc_bytes": + return &m.NextGC + case "go_goroutines": + return &m.Goroutines + case "process_resident_memory_bytes": + return &m.ProcessRSS + } + return nil +} + +// parseMetrics reads the Prometheus text exposition format far enough to pull +// out the handful of unlabelled gauges above. It deliberately skips any sample +// carrying labels: every metric this harness wants is a single series, so a +// name with a "{" is one of objgitd's own vectors and not a match. +func parseMetrics(r io.Reader) (goMetrics, error) { + var m goMetrics + + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 0, 64*1024), 1<<20) + for sc.Scan() { + line := sc.Text() + if line == "" || line[0] == '#' { + continue + } + + name, value, ok := strings.Cut(line, " ") + if !ok || strings.ContainsRune(name, '{') { + continue + } + + dst := m.field(name) + if dst == nil { + continue + } + + f, err := strconv.ParseFloat(value, 64) + if err != nil { + return m, fmt.Errorf("membench: metric %s has non-numeric value %q: %w", name, value, err) + } + if f < 0 { + f = 0 + } + *dst = uint64(f) + } + if err := sc.Err(); err != nil { + return m, fmt.Errorf("membench: reading metrics: %w", err) + } + + return m, nil +} diff --git a/cmd/membench/metrics_test.go b/cmd/membench/metrics_test.go new file mode 100644 index 0000000..bba6d75 --- /dev/null +++ b/cmd/membench/metrics_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "strings" + "testing" +) + +func TestParseMetrics(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + input string + want goMetrics + wantErr bool + }{ + { + name: "the metrics the harness records", + input: "# HELP go_goroutines Number of goroutines that currently exist.\n" + + "# TYPE go_goroutines gauge\n" + + "go_goroutines 42\n" + + "go_memstats_heap_inuse_bytes 1.2345678e+07\n" + + "go_memstats_heap_sys_bytes 3.3554432e+07\n" + + "go_memstats_next_gc_bytes 4.194304e+06\n" + + "process_resident_memory_bytes 2.8835840e+07\n", + want: goMetrics{ + HeapInuse: 12345678, + HeapSys: 33554432, + NextGC: 4194304, + Goroutines: 42, + ProcessRSS: 28835840, + }, + }, + { + name: "labelled series are skipped", + input: "objgit_pushes_total{repo=\"a/b\"} 99\n" + + "go_goroutines 7\n", + want: goMetrics{Goroutines: 7}, + }, + { + name: "a metric name that happens to prefix a wanted one", + input: "go_goroutines_extra 5\n", + want: goMetrics{}, + }, + { + name: "non-numeric value", + input: "go_goroutines NaNsense\n", + wantErr: true, + }, + { + name: "comments and blanks only", + input: "# HELP nothing\n\n", + want: goMetrics{}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := parseMetrics(strings.NewReader(tt.input)) + if tt.wantErr { + if err == nil { + t.Error("wanted an error, got none") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if got != tt.want { + t.Logf("want: %+v", tt.want) + t.Logf("got: %+v", got) + t.Error("parsed the wrong metrics") + } + }) + } +} diff --git a/cmd/membench/proc.go b/cmd/membench/proc.go new file mode 100644 index 0000000..1b982e4 --- /dev/null +++ b/cmd/membench/proc.go @@ -0,0 +1,83 @@ +package main + +import ( + "bufio" + "fmt" + "io" + "os" + "strconv" + "strings" +) + +// procStatus is the subset of /proc//status and /proc//smaps_rollup +// this harness records, in bytes. The kernel reports kB; parseProcKV converts. +// +// VmHWM is the number that matters for sizing: it is the high-water mark the +// kernel itself keeps, so it cannot be missed between two samples the way a +// spike in VmRSS can be. +type procStatus struct { + VmRSS uint64 + VmHWM uint64 + Pss uint64 + PrivateDirty uint64 +} + +// parseProcKV reads the "Key:\t 1234 kB" lines that both /proc//status +// and /proc//smaps_rollup use, and returns them in bytes. Lines that do +// not carry a kB count (Name, State, the smaps_rollup header) are skipped, +// because no caller wants them. +func parseProcKV(r io.Reader) (map[string]uint64, error) { + out := map[string]uint64{} + sc := bufio.NewScanner(r) + for sc.Scan() { + key, rest, ok := strings.Cut(sc.Text(), ":") + if !ok { + continue + } + fields := strings.Fields(rest) + if len(fields) != 2 || fields[1] != "kB" { + continue + } + n, err := strconv.ParseUint(fields[0], 10, 64) + if err != nil { + return nil, fmt.Errorf("membench: %q is not a kB count: %w", sc.Text(), err) + } + out[key] = n * 1024 + } + if err := sc.Err(); err != nil { + return nil, fmt.Errorf("membench: reading proc file: %w", err) + } + return out, nil +} + +// readProcStatus samples one process. /proc//status is required; +// smaps_rollup is best effort, because it needs a 4.14 or newer kernel and +// permission to read another process's mappings. Losing Pss costs detail, not +// the measurement. +func readProcStatus(pid int) (procStatus, error) { + var ps procStatus + + kv, err := readProcFile(fmt.Sprintf("/proc/%d/status", pid)) + if err != nil { + return ps, err + } + ps.VmRSS = kv["VmRSS"] + ps.VmHWM = kv["VmHWM"] + + if rollup, err := readProcFile(fmt.Sprintf("/proc/%d/smaps_rollup", pid)); err == nil { + ps.Pss = rollup["Pss"] + ps.PrivateDirty = rollup["Private_Dirty"] + } + + return ps, nil +} + +func readProcFile(name string) (map[string]uint64, error) { + f, err := os.Open(name) + if err != nil { + return nil, fmt.Errorf("membench: can't open %s: %w", name, err) + } + defer f.Close() + + return parseProcKV(f) +} diff --git a/cmd/membench/proc_test.go b/cmd/membench/proc_test.go new file mode 100644 index 0000000..2a89917 --- /dev/null +++ b/cmd/membench/proc_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "strings" + "testing" +) + +func TestParseProcKV(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + input string + want map[string]uint64 + wantErr bool + }{ + { + name: "status excerpt", + input: "Name:\tobjgitd\n" + + "State:\tS (sleeping)\n" + + "VmRSS:\t 28160 kB\n" + + "VmHWM:\t 31744 kB\n", + want: map[string]uint64{"VmRSS": 28160 * 1024, "VmHWM": 31744 * 1024}, + }, + { + name: "smaps_rollup excerpt", + input: "55d0c0000000-7ffd0c0f1000 ---p 00000000 00:00 0 [rollup]\n" + + "Rss: 7168 kB\n" + + "Pss: 4096 kB\n" + + "Private_Dirty: 2048 kB\n", + want: map[string]uint64{"Rss": 7168 * 1024, "Pss": 4096 * 1024, "Private_Dirty": 2048 * 1024}, + }, + { + name: "no kB lines", + input: "Name:\tobjgitd\nThreads:\t12\n", + want: map[string]uint64{}, + }, + { + name: "unparseable count", + input: "VmRSS:\t twelve kB\n", + wantErr: true, + }, + { + name: "empty", + input: "", + want: map[string]uint64{}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := parseProcKV(strings.NewReader(tt.input)) + if tt.wantErr { + if err == nil { + t.Error("wanted an error, got none") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(got) != len(tt.want) { + t.Logf("want: %v", tt.want) + t.Logf("got: %v", got) + t.Fatalf("got %d keys, want %d", len(got), len(tt.want)) + } + for k, want := range tt.want { + if got[k] != want { + t.Logf("want: %d", want) + t.Logf("got: %d", got[k]) + t.Errorf("%s is wrong", k) + } + } + }) + } +} diff --git a/cmd/membench/report.go b/cmd/membench/report.go new file mode 100644 index 0000000..3a6b5a6 --- /dev/null +++ b/cmd/membench/report.go @@ -0,0 +1,374 @@ +package main + +import ( + "encoding/csv" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "time" +) + +// pushResult is one git push the harness drove, with the window the sampler +// should be searched over to find what that push cost. +type pushResult struct { + Phase string + Concurrency int + Repo string + Start time.Time + End time.Time + Err error +} + +func (p pushResult) duration() time.Duration { return p.End.Sub(p.Start) } + +// runMeta is everything about the run that is not a measurement, but that the +// numbers are meaningless without. Go's resident set is a function of GOGC and +// GOMEMLIMIT, so a peak-RSS figure recorded without them cannot be compared to +// anything. +type runMeta struct { + StartedAt time.Time + FinishedAt time.Time + Host string + GoVersion string + GOGC string + GOMEMLIMIT string + NumCPU int + SourceRepo string + PackBytes int64 + Org string + DaemonArgs []string + SampleEvery time.Duration + WindowSlack time.Duration + RunDir string +} + +// windowPeak returns the highest resident set and heap-in-use seen between two +// instants. The window is widened by one sample interval on each side because +// a push's cost does not land inside its own wall clock exactly: the daemon is +// still flushing to the bucket when git has already exited. +func windowPeak(samples []sample, start, end time.Time, slack time.Duration) (rss, heap uint64) { + from, to := start.Add(-slack), end.Add(slack) + for _, s := range samples { + if s.At.Before(from) || s.At.After(to) { + continue + } + if s.Proc.VmRSS > rss { + rss = s.Proc.VmRSS + } + if s.GoOK && s.Go.HeapInuse > heap { + heap = s.Go.HeapInuse + } + } + return rss, heap +} + +// rssAt returns the resident set at the most recent sample at or before t, and +// zero when the sampler had not started yet. It is what makes retention legible: +// the reading going into a push, compared with the reading coming out of it. +func rssAt(samples []sample, t time.Time) uint64 { + var out uint64 + for _, s := range samples { + if s.At.After(t) { + break + } + out = s.Proc.VmRSS + } + return out +} + +// phaseTail returns the last sample recorded for a phase, which is the settled +// reading for it: the idle value after everything has drained. +// +// Samples whose metrics scrape failed are skipped. The last tick before the +// daemon exits routinely fails, and taking it would report a settled heap of +// zero rather than the figure that was actually reached. +func phaseTail(samples []sample, phase string) (sample, bool) { + for i := len(samples) - 1; i >= 0; i-- { + if samples[i].Phase == phase && samples[i].GoOK { + return samples[i], true + } + } + return sample{}, false +} + +// goCell renders a runtime metric, or an empty cell when that tick's scrape +// failed. Empty is deliberate: a zero here would plot as a cliff to zero. +func goCell(ok bool, v uint64) string { + if !ok { + return "" + } + return strconv.FormatUint(v, 10) +} + +func writeCSV(path string, samples []sample) error { + f, err := os.Create(path) + if err != nil { + return fmt.Errorf("membench: can't create %s: %w", path, err) + } + defer f.Close() + + w := csv.NewWriter(f) + defer w.Flush() + + header := []string{ + "t_ms", "iso8601", "phase", "repo", + "vm_rss", "vm_hwm", "pss", "private_dirty", + "heap_inuse", "heap_sys", "next_gc", "goroutines", "process_rss", + } + if err := w.Write(header); err != nil { + return fmt.Errorf("membench: can't write CSV header: %w", err) + } + + if len(samples) == 0 { + return nil + } + + origin := samples[0].At + for _, s := range samples { + row := []string{ + strconv.FormatInt(s.At.Sub(origin).Milliseconds(), 10), + s.At.Format(time.RFC3339Nano), + s.Phase, + s.Repo, + strconv.FormatUint(s.Proc.VmRSS, 10), + strconv.FormatUint(s.Proc.VmHWM, 10), + strconv.FormatUint(s.Proc.Pss, 10), + strconv.FormatUint(s.Proc.PrivateDirty, 10), + goCell(s.GoOK, s.Go.HeapInuse), + goCell(s.GoOK, s.Go.HeapSys), + goCell(s.GoOK, s.Go.NextGC), + goCell(s.GoOK, s.Go.Goroutines), + goCell(s.GoOK, s.Go.ProcessRSS), + } + if err := w.Write(row); err != nil { + return fmt.Errorf("membench: can't write CSV row: %w", err) + } + } + + return w.Error() +} + +// mib formats bytes as MiB, which is the only unit anyone reads these numbers +// in. +func mib(b uint64) string { return fmt.Sprintf("%.1f MiB", float64(b)/(1<<20)) } + +// deltaMiB formats a signed difference in MiB. +func deltaMiB(now, base uint64) string { + d := (float64(now) - float64(base)) / (1 << 20) + return fmt.Sprintf("%+.1f MiB", d) +} + +func writeReport(path string, meta runMeta, samples []sample, profiles []profileCapture, pushes []pushResult) error { + var b strings.Builder + + baseRSS, baseHeap := uint64(0), uint64(0) + if s, ok := phaseTail(samples, phaseBaseline); ok { + baseRSS, baseHeap = s.Proc.VmRSS, s.Go.HeapInuse + } + + var hwm uint64 + for _, s := range samples { + if s.Proc.VmHWM > hwm { + hwm = s.Proc.VmHWM + } + } + + fmt.Fprintf(&b, "# objgitd push memory benchmark\n\n") + fmt.Fprintf(&b, "Run started %s, finished %s (%s).\n\n", + meta.StartedAt.Format(time.RFC3339), + meta.FinishedAt.Format(time.RFC3339), + meta.FinishedAt.Sub(meta.StartedAt).Round(time.Second)) + + fmt.Fprintf(&b, "## Headline\n\n") + fmt.Fprintf(&b, "- Peak resident set for the whole run (VmHWM): **%s**\n", mib(hwm)) + fmt.Fprintf(&b, "- Idle baseline before any push: %s resident, %s heap in use\n", mib(baseRSS), mib(baseHeap)) + fmt.Fprintf(&b, "- Source pack pushed each time: %s\n", mib(uint64(meta.PackBytes))) + if s, ok := phaseTail(samples, phaseIdle); ok { + fmt.Fprintf(&b, "- Settled after the last push, once idle: %s resident, %s heap in use (%s of heap over baseline)\n", + mib(s.Proc.VmRSS), mib(s.Go.HeapInuse), deltaMiB(s.Go.HeapInuse, baseHeap)) + } + fmt.Fprintf(&b, "- Repositories created: %d under `%s/`\n\n", len(pushes), meta.Org) + + fmt.Fprintf(&b, "## Run conditions\n\n") + fmt.Fprintf(&b, "| Setting | Value |\n| --- | --- |\n") + fmt.Fprintf(&b, "| Host | %s |\n", meta.Host) + fmt.Fprintf(&b, "| Go | %s |\n", meta.GoVersion) + fmt.Fprintf(&b, "| GOGC | %s |\n", orDefault(meta.GOGC, "unset (100)")) + fmt.Fprintf(&b, "| GOMEMLIMIT | %s |\n", orDefault(meta.GOMEMLIMIT, "unset (no limit)")) + fmt.Fprintf(&b, "| CPUs | %d |\n", meta.NumCPU) + fmt.Fprintf(&b, "| Source repo | %s |\n", meta.SourceRepo) + fmt.Fprintf(&b, "| Sample interval | %s |\n", meta.SampleEvery) + fmt.Fprintf(&b, "| Daemon flags | `%s` |\n", strings.Join(meta.DaemonArgs, " ")) + fmt.Fprintf(&b, "| Run directory | %s |\n\n", meta.RunDir) + + slack := meta.WindowSlack + + fmt.Fprintf(&b, "## Sequential pushes\n\n") + fmt.Fprintf(&b, "One push at a time, each to a fresh repository, with an idle gap between them. ") + fmt.Fprintf(&b, "Peaks are taken over the push plus %s of slack, so work the daemon finishes after git exits still counts against the push that caused it.\n\n", slack) + fmt.Fprintf(&b, "Read the before and after columns together. After is the settled reading once the idle gap has passed: if it tracks upwards push after push instead of falling back to before, the daemon is keeping memory it no longer needs.\n\n") + fmt.Fprintf(&b, "| # | Repo | Wall clock | RSS before | Peak RSS | Peak heap | RSS after | Kept | Result |\n") + fmt.Fprintf(&b, "| --- | --- | --- | --- | --- | --- | --- | --- | --- |\n") + n := 0 + for _, p := range pushes { + if p.Phase != phaseSeq { + continue + } + n++ + rss, heap := windowPeak(samples, p.Start, p.End, slack) + before := rssAt(samples, p.Start) + after := rssAt(samples, p.End.Add(slack)) + fmt.Fprintf(&b, "| %d | `%s` | %s | %s | %s | %s | %s | %s | %s |\n", + n, p.Repo, p.duration().Round(time.Millisecond), + mib(before), mib(rss), mib(heap), mib(after), deltaMiB(after, before), result(p.Err)) + } + + fmt.Fprintf(&b, "\n## Concurrency sweep\n\n") + fmt.Fprintf(&b, "K simultaneous pushes to K fresh repositories. The rise is measured from the reading going into the step, not from the original idle baseline: ") + fmt.Fprintf(&b, "if earlier pushes left memory behind, measuring from the baseline would charge that to this step as well.\n\n") + fmt.Fprintf(&b, "Per concurrent push is the rise divided by K. That is the slope to size a machine with, and it is only meaningful if it stays flat as K grows.\n\n") + fmt.Fprintf(&b, "| K | Wall clock | RSS before | Peak RSS | Peak heap | Rise | Per concurrent push | Failures |\n") + fmt.Fprintf(&b, "| --- | --- | --- | --- | --- | --- | --- | --- |\n") + for _, step := range concurrencySteps(pushes) { + group := pushesAt(pushes, step) + start, end := groupWindow(group) + rss, heap := windowPeak(samples, start, end, slack) + before := rssAt(samples, start) + rise := float64(rss) - float64(before) + fails := 0 + for _, p := range group { + if p.Err != nil { + fails++ + } + } + fmt.Fprintf(&b, "| %d | %s | %s | %s | %s | %s | %.1f MiB | %d |\n", + step, end.Sub(start).Round(time.Millisecond), mib(before), mib(rss), mib(heap), + deltaMiB(rss, before), rise/float64(step)/(1<<20), fails) + } + + fmt.Fprintf(&b, "\n## Captured profiles\n\n") + if len(profiles) == 0 { + fmt.Fprintf(&b, "None.\n") + } else { + fmt.Fprintf(&b, "| File | Taken at | RSS then | Why |\n| --- | --- | --- | --- |\n") + for _, p := range profiles { + fmt.Fprintf(&b, "| `%s` | %s | %s | %s |\n", + p.Name, p.At.Format("15:04:05.000"), mib(p.RSS), p.Reason) + } + } + + fmt.Fprintf(&b, "\n## Reading the profiles\n\n") + fmt.Fprintf(&b, "```sh\n") + fmt.Fprintf(&b, "# What was on the heap at the worst moment.\n") + fmt.Fprintf(&b, "go tool pprof -http=: %s/heap-peak-*.pb.gz\n\n", meta.RunDir) + fmt.Fprintf(&b, "# What the pushes left behind: the settled heap minus the idle baseline.\n") + fmt.Fprintf(&b, "go tool pprof -http=: -base %s/heap-baseline.pb.gz %s/heap-final.pb.gz\n\n", meta.RunDir, meta.RunDir) + fmt.Fprintf(&b, "# Total allocation over the whole run, which finds the churn GC is working to keep up with.\n") + fmt.Fprintf(&b, "go tool pprof -sample_index=alloc_space -http=: %s/allocs-final.pb.gz\n", meta.RunDir) + fmt.Fprintf(&b, "```\n\n") + + fmt.Fprintf(&b, "## Caveats\n\n") + fmt.Fprintf(&b, "- Peak heap profiles are taken with `gc=0`. Forcing a collection would change the number that triggered the capture, so the profile shows the heap as it was, sampling error included.\n") + fmt.Fprintf(&b, "- Resident set lags the heap. Go returns freed pages to the kernel lazily, so RSS staying high after a push is not by itself a leak; the baseline-diffed heap profile is what settles that.\n") + fmt.Fprintf(&b, "- Pushes go to a real Tigris bucket, so wall clock includes network time and varies run to run. Memory does not depend on it, but throughput comparisons across runs do.\n") + fmt.Fprintf(&b, "- The pack cache is on local disk (`-pack-cache-dir` under the run directory) and is cold at the start of every run. Disk pressure there is not memory pressure.\n") + + if err := os.WriteFile(path, []byte(b.String()), 0o644); err != nil { + return fmt.Errorf("membench: can't write %s: %w", path, err) + } + + return nil +} + +func orDefault(s, fallback string) string { + if s == "" { + return fallback + } + return s +} + +func result(err error) string { + if err != nil { + return "FAILED: " + err.Error() + } + return "ok" +} + +// concurrencySteps returns the distinct K values in the concurrency phase, in +// ascending order. +func concurrencySteps(pushes []pushResult) []int { + seen := map[int]bool{} + var out []int + for _, p := range pushes { + if p.Phase != phaseConc || seen[p.Concurrency] { + continue + } + seen[p.Concurrency] = true + out = append(out, p.Concurrency) + } + sort.Ints(out) + return out +} + +func pushesAt(pushes []pushResult, k int) []pushResult { + var out []pushResult + for _, p := range pushes { + if p.Phase == phaseConc && p.Concurrency == k { + out = append(out, p) + } + } + return out +} + +// groupWindow is the span from the first push starting to the last one +// finishing, which is the window the whole concurrent step occupied. +func groupWindow(group []pushResult) (start, end time.Time) { + for i, p := range group { + if i == 0 || p.Start.Before(start) { + start = p.Start + } + if i == 0 || p.End.After(end) { + end = p.End + } + } + return start, end +} + +// writeRepoList records every repository the run created, so the bucket can be +// cleaned up afterwards. The harness never deletes anything itself. +func writeRepoList(path string, org string, pushes []pushResult) error { + var b strings.Builder + for _, p := range pushes { + fmt.Fprintf(&b, "%s/%s\n", org, p.Repo) + } + if err := os.WriteFile(path, []byte(b.String()), 0o644); err != nil { + return fmt.Errorf("membench: can't write %s: %w", path, err) + } + return nil +} + +// packBytes totals the pack files in a bare repository, which is how much data +// each push actually moves. +func packBytes(gitDir string) (int64, error) { + entries, err := os.ReadDir(filepath.Join(gitDir, "objects", "pack")) + if err != nil { + return 0, fmt.Errorf("membench: can't read pack directory: %w", err) + } + + var total int64 + for _, e := range entries { + if !strings.HasSuffix(e.Name(), ".pack") { + continue + } + info, err := e.Info() + if err != nil { + return 0, fmt.Errorf("membench: can't stat %s: %w", e.Name(), err) + } + total += info.Size() + } + + return total, nil +} diff --git a/cmd/membench/report_test.go b/cmd/membench/report_test.go new file mode 100644 index 0000000..1c6ac37 --- /dev/null +++ b/cmd/membench/report_test.go @@ -0,0 +1,313 @@ +package main + +import ( + "regexp" + "testing" + "time" +) + +// at builds a sample n seconds after a fixed origin, so window arithmetic in +// these tests reads as plainly as it can. +func at(origin time.Time, sec int, phase string, rss, heap uint64) sample { + return sample{ + At: origin.Add(time.Duration(sec) * time.Second), + Phase: phase, + Proc: procStatus{VmRSS: rss, VmHWM: rss}, + Go: goMetrics{HeapInuse: heap}, + GoOK: true, + } +} + +// atFailedScrape is a tick where /proc was read but the metrics endpoint did +// not answer, which is what the last sample before shutdown looks like. +func atFailedScrape(origin time.Time, sec int, phase string, rss uint64) sample { + s := at(origin, sec, phase, rss, 0) + s.Go = goMetrics{} + s.GoOK = false + return s +} + +func TestPhaseTailSkipsFailedScrapes(t *testing.T) { + t.Parallel() + + origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + samples := []sample{ + at(origin, 0, phaseIdle, 700, 500), + atFailedScrape(origin, 1, phaseIdle, 690), + } + + got, ok := phaseTail(samples, phaseIdle) + if !ok { + t.Fatal("wanted the last good sample, got none") + } + if got.Go.HeapInuse != 500 { + t.Logf("want: 500") + t.Logf("got: %d", got.Go.HeapInuse) + t.Error("a failed scrape was reported as a settled heap") + } +} + +func TestWindowPeakIgnoresFailedScrapeHeap(t *testing.T) { + t.Parallel() + + origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + samples := []sample{ + at(origin, 1, phaseSeq, 100, 60), + atFailedScrape(origin, 2, phaseSeq, 900), + } + + rss, heap := windowPeak(samples, origin.Add(time.Second), origin.Add(2*time.Second), 0) + if rss != 900 { + t.Logf("want rss: 900") + t.Logf("got rss: %d", rss) + t.Error("proc numbers should survive a failed scrape") + } + if heap != 60 { + t.Logf("want heap: 60") + t.Logf("got heap: %d", heap) + t.Error("heap from a failed scrape leaked into the peak") + } +} + +func TestWindowPeak(t *testing.T) { + t.Parallel() + + origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + samples := []sample{ + at(origin, 0, phaseBaseline, 10, 5), + at(origin, 1, phaseSeq, 40, 20), + at(origin, 2, phaseSeq, 90, 60), + at(origin, 3, phaseIdle, 30, 10), + at(origin, 9, phaseSeq, 999, 999), + } + + for _, tt := range []struct { + name string + start, end int + slack time.Duration + wantRSS uint64 + wantHeap uint64 + }{ + {name: "covers the push window", start: 1, end: 2, wantRSS: 90, wantHeap: 60}, + {name: "slack pulls in the settling sample", start: 1, end: 2, slack: time.Second, wantRSS: 90, wantHeap: 60}, + {name: "excludes a later unrelated spike", start: 0, end: 3, wantRSS: 90, wantHeap: 60}, + {name: "empty window", start: 5, end: 6, wantRSS: 0, wantHeap: 0}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + start := origin.Add(time.Duration(tt.start) * time.Second) + end := origin.Add(time.Duration(tt.end) * time.Second) + + rss, heap := windowPeak(samples, start, end, tt.slack) + if rss != tt.wantRSS || heap != tt.wantHeap { + t.Logf("want: rss=%d heap=%d", tt.wantRSS, tt.wantHeap) + t.Logf("got: rss=%d heap=%d", rss, heap) + t.Error("wrong peak") + } + }) + } +} + +func TestRSSAt(t *testing.T) { + t.Parallel() + + origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + samples := []sample{ + at(origin, 1, phaseBaseline, 10, 5), + at(origin, 2, phaseSeq, 90, 60), + at(origin, 3, phaseIdle, 70, 40), + } + + for _, tt := range []struct { + name string + sec int + want uint64 + }{ + {name: "before the first sample", sec: 0, want: 0}, + {name: "exactly on a sample", sec: 2, want: 90}, + {name: "between samples takes the earlier one", sec: 2, want: 90}, + {name: "after the last sample", sec: 99, want: 70}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got := rssAt(samples, origin.Add(time.Duration(tt.sec)*time.Second)) + if got != tt.want { + t.Logf("want: %d", tt.want) + t.Logf("got: %d", got) + t.Error("wrong reading") + } + }) + } +} + +func TestPhaseTail(t *testing.T) { + t.Parallel() + + origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + samples := []sample{ + at(origin, 0, phaseBaseline, 10, 5), + at(origin, 1, phaseBaseline, 12, 6), + at(origin, 2, phaseSeq, 90, 60), + } + + for _, tt := range []struct { + name string + phase string + wantRSS uint64 + wantOK bool + }{ + {name: "last baseline sample wins", phase: phaseBaseline, wantRSS: 12, wantOK: true}, + {name: "only sample in the phase", phase: phaseSeq, wantRSS: 90, wantOK: true}, + {name: "phase never happened", phase: phaseConc, wantOK: false}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, ok := phaseTail(samples, tt.phase) + if ok != tt.wantOK { + t.Logf("want ok: %v", tt.wantOK) + t.Logf("got ok: %v", ok) + t.Fatal("wrong presence") + } + if ok && got.Proc.VmRSS != tt.wantRSS { + t.Logf("want: %d", tt.wantRSS) + t.Logf("got: %d", got.Proc.VmRSS) + t.Error("wrong sample") + } + }) + } +} + +func TestConcurrencySteps(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + pushes []pushResult + want []int + }{ + { + name: "distinct levels, ascending, sequential pushes ignored", + pushes: []pushResult{ + {Phase: phaseSeq, Concurrency: 1}, + {Phase: phaseConc, Concurrency: 4}, + {Phase: phaseConc, Concurrency: 4}, + {Phase: phaseConc, Concurrency: 1}, + {Phase: phaseConc, Concurrency: 2}, + }, + want: []int{1, 2, 4}, + }, + {name: "no concurrency phase", pushes: []pushResult{{Phase: phaseSeq, Concurrency: 1}}}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got := concurrencySteps(tt.pushes) + if len(got) != len(tt.want) { + t.Logf("want: %v", tt.want) + t.Logf("got: %v", got) + t.Fatal("wrong number of steps") + } + for i := range got { + if got[i] != tt.want[i] { + t.Logf("want: %v", tt.want) + t.Logf("got: %v", got) + t.Fatal("wrong steps") + } + } + }) + } +} + +func TestGroupWindow(t *testing.T) { + t.Parallel() + + origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + group := []pushResult{ + {Start: origin.Add(2 * time.Second), End: origin.Add(5 * time.Second)}, + {Start: origin, End: origin.Add(3 * time.Second)}, + {Start: origin.Add(time.Second), End: origin.Add(9 * time.Second)}, + } + + start, end := groupWindow(group) + if !start.Equal(origin) { + t.Logf("want: %v", origin) + t.Logf("got: %v", start) + t.Error("wrong window start") + } + if want := origin.Add(9 * time.Second); !end.Equal(want) { + t.Logf("want: %v", want) + t.Logf("got: %v", end) + t.Error("wrong window end") + } +} + +func TestParseSteps(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + input string + want []int + wantErr bool + }{ + {name: "the default sweep", input: "1,2,4,8", want: []int{1, 2, 4, 8}}, + {name: "spaces and a trailing comma", input: " 1, 3 ,", want: []int{1, 3}}, + {name: "empty disables the sweep", input: " "}, + {name: "not a number", input: "1,two", wantErr: true}, + {name: "zero is not a concurrency", input: "0", wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := parseSteps(tt.input) + if tt.wantErr { + if err == nil { + t.Error("wanted an error, got none") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(got) != len(tt.want) { + t.Logf("want: %v", tt.want) + t.Logf("got: %v", got) + t.Fatal("wrong number of steps") + } + for i := range got { + if got[i] != tt.want[i] { + t.Logf("want: %v", tt.want) + t.Logf("got: %v", got) + t.Fatal("wrong steps") + } + } + }) + } +} + +func TestNewUUID(t *testing.T) { + t.Parallel() + + shape := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`) + + seen := map[string]bool{} + for range 100 { + got, err := newUUID() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !shape.MatchString(got) { + t.Logf("want: a version 4 UUID") + t.Logf("got: %s", got) + t.Fatal("wrong shape") + } + if seen[got] { + t.Fatalf("%s came back twice", got) + } + seen[got] = true + } +} diff --git a/cmd/membench/sampler.go b/cmd/membench/sampler.go new file mode 100644 index 0000000..f93cde7 --- /dev/null +++ b/cmd/membench/sampler.go @@ -0,0 +1,224 @@ +package main + +import ( + "context" + "fmt" + "io" + "log/slog" + "net/http" + "os" + "path/filepath" + "sync" + "time" +) + +// sample is one row of samples.csv: everything the harness knew about the +// daemon at one instant, tagged with what the daemon was being asked to do. +type sample struct { + At time.Time + Phase string + Repo string + Proc procStatus + Go goMetrics + // GoOK is false when the /metrics scrape failed for this tick, which + // happens routinely while the daemon is shutting down. The proc numbers are + // still good, so the row is kept, but every Go figure in it is a zero value + // rather than a measurement and must not be read as one. + GoOK bool +} + +// profileCapture records one pprof profile written to disk, so the report can +// say which resident-memory reading each profile belongs to. +type profileCapture struct { + Name string + Path string + At time.Time + RSS uint64 + Reason string +} + +// sampler polls the daemon's /proc entries and /metrics on a fixed tick, and +// grabs a heap profile whenever resident memory sets a new high-water mark. +// +// The phase and repo labels are set by the driver as it walks the run, so every +// row says what the daemon was doing when it was taken. That is what turns the +// CSV from a wall of numbers into something you can attribute. +type sampler struct { + pid int + metricsURL string + pprofURL string + outDir string + interval time.Duration + growth float64 + cooldown time.Duration + client *http.Client + + mu sync.Mutex + phase string + repo string + samples []sample + profiles []profileCapture + + peakRSS uint64 + lastCapture time.Time +} + +func newSampler(pid int, metricsAddr, outDir string, interval, cooldown time.Duration, growth float64) *sampler { + base := "http://" + metricsAddr + return &sampler{ + pid: pid, + metricsURL: base + "/metrics", + pprofURL: base + "/debug/pprof", + outDir: outDir, + interval: interval, + growth: growth, + cooldown: cooldown, + client: &http.Client{Timeout: 30 * time.Second}, + } +} + +// label tells the sampler what the driver is about to do. Every sample taken +// until the next call carries these two values. +func (s *sampler) label(phase, repo string) { + s.mu.Lock() + defer s.mu.Unlock() + s.phase, s.repo = phase, repo +} + +// run samples until the context is cancelled. +func (s *sampler) run(ctx context.Context) { + t := time.NewTicker(s.interval) + defer t.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.collect(ctx) + } + } +} + +// collect takes one sample. A failed read is logged and dropped rather than +// aborting the run: a single missed tick costs one row, and the benchmark is +// several minutes of pushes we do not want to throw away. +func (s *sampler) collect(ctx context.Context) { + ps, err := readProcStatus(s.pid) + if err != nil { + slog.Debug("can't read proc status", "pid", s.pid, "err", err) + return + } + + gm, err := s.fetchMetrics(ctx) + goOK := err == nil + if err != nil { + slog.Debug("can't read metrics", "url", s.metricsURL, "err", err) + } + + s.mu.Lock() + now := time.Now() + s.samples = append(s.samples, sample{At: now, Phase: s.phase, Repo: s.repo, Proc: ps, Go: gm, GoOK: goOK}) + + capture := false + if float64(ps.VmRSS) > float64(s.peakRSS)*(1+s.growth) && now.Sub(s.lastCapture) > s.cooldown { + capture = true + s.lastCapture = now + } + if ps.VmRSS > s.peakRSS { + s.peakRSS = ps.VmRSS + } + phase, repo := s.phase, s.repo + s.mu.Unlock() + + if !capture { + return + } + + // gc=0 on purpose. Forcing a collection here would change the very number + // that triggered the capture; we want the heap as it actually was at the + // peak, sampling error and all. + name := fmt.Sprintf("heap-peak-%s-%d.pb.gz", now.Format("150405.000"), ps.VmRSS) + reason := fmt.Sprintf("new RSS high-water mark during %s %s", phase, repo) + if _, err := s.captureProfile(ctx, "heap", false, name, reason, ps.VmRSS); err != nil { + slog.Warn("can't capture peak heap profile", "err", err) + } +} + +func (s *sampler) fetchMetrics(ctx context.Context) (goMetrics, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.metricsURL, nil) + if err != nil { + return goMetrics{}, fmt.Errorf("membench: can't build metrics request: %w", err) + } + + resp, err := s.client.Do(req) + if err != nil { + return goMetrics{}, fmt.Errorf("membench: metrics request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return goMetrics{}, fmt.Errorf("membench: %s returned %s", s.metricsURL, resp.Status) + } + + return parseMetrics(resp.Body) +} + +// captureProfile fetches one pprof profile and writes it into the run +// directory. gc forces a collection before the heap is dumped: use it for the +// baseline and the final capture, where a settled heap is what you want, and +// never mid-push. +func (s *sampler) captureProfile(ctx context.Context, kind string, gc bool, name, reason string, rss uint64) (profileCapture, error) { + url := fmt.Sprintf("%s/%s?gc=0", s.pprofURL, kind) + if gc { + url = fmt.Sprintf("%s/%s?gc=1", s.pprofURL, kind) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return profileCapture{}, fmt.Errorf("membench: can't build pprof request: %w", err) + } + + resp, err := s.client.Do(req) + if err != nil { + return profileCapture{}, fmt.Errorf("membench: pprof request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return profileCapture{}, fmt.Errorf("membench: %s returned %s", url, resp.Status) + } + + path := filepath.Join(s.outDir, name) + f, err := os.Create(path) + if err != nil { + return profileCapture{}, fmt.Errorf("membench: can't create %s: %w", path, err) + } + defer f.Close() + + if _, err := io.Copy(f, resp.Body); err != nil { + return profileCapture{}, fmt.Errorf("membench: can't write %s: %w", path, err) + } + + pc := profileCapture{Name: name, Path: path, At: time.Now(), RSS: rss, Reason: reason} + s.mu.Lock() + s.profiles = append(s.profiles, pc) + s.mu.Unlock() + + slog.Info("captured profile", "name", name, "reason", reason, "rss_bytes", rss) + return pc, nil +} + +// snapshot returns copies of everything collected so far, so the report writer +// never touches the sampler's state while it is still running. +func (s *sampler) snapshot() ([]sample, []profileCapture) { + s.mu.Lock() + defer s.mu.Unlock() + + samples := make([]sample, len(s.samples)) + copy(samples, s.samples) + profiles := make([]profileCapture, len(s.profiles)) + copy(profiles, s.profiles) + + return samples, profiles +} diff --git a/cmd/membench/uuid.go b/cmd/membench/uuid.go new file mode 100644 index 0000000..aaa7df4 --- /dev/null +++ b/cmd/membench/uuid.go @@ -0,0 +1,21 @@ +package main + +import ( + "crypto/rand" + "fmt" +) + +// newUUID returns a random version 4 UUID. The harness only needs repository +// names that never collide with an earlier run, so crypto/rand covers it and no +// dependency is needed. +func newUUID() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", fmt.Errorf("membench: can't read random bytes: %w", err) + } + + b[6] = (b[6] & 0x0f) | 0x40 // version 4 + b[8] = (b[8] & 0x3f) | 0x80 // variant 10 + + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]), nil +} diff --git a/docs/usage/memory-benchmark.md b/docs/usage/memory-benchmark.md new file mode 100644 index 0000000..b725271 --- /dev/null +++ b/docs/usage/memory-benchmark.md @@ -0,0 +1,124 @@ +# Measuring push memory + +`cmd/membench` answers one question: how much memory does `objgitd` need while +it is taking pushes, and does that memory come back afterwards. + +It starts a daemon it owns, pushes one real repository into many fresh ones, +samples the daemon's resident set and Go heap throughout, and captures a pprof +heap profile every time memory sets a new high-water mark. Nothing in `objgitd` +had to change for this: `cmd/objgitd/main.go` already serves `net/http/pprof` +and the default Prometheus registry (which carries `go_memstats_*` and +`process_resident_memory_bytes`) on the metrics listener. + +## Running it + +```text +go run ./cmd/membench -out ./bench-out +``` + +Run it from the objgit checkout. The harness finds `go.mod`, builds +`./cmd/objgitd`, and starts the daemon with that directory as its working +directory, so the daemon picks up `BUCKET` and the AWS credentials from `.env` +exactly as it would normally. + +Pushes go to a real Tigris bucket. There is no local S3 fake in this repo, and a +fake would hide the allocation behaviour of the Tigris storer, which is the +thing being measured. + +## What a run does + +1. **Mirror.** `git clone --mirror` of the source repository into the run + directory, once. Every push reads from that copy, so the repository you + pointed at is never written to and concurrent pushes share an immutable + source. +2. **Baseline.** Sample the idle daemon for `-baseline`, then capture + `heap-baseline.pb.gz` with a forced GC. Every later number is read against + this. +3. **Sequential.** `-seq-pushes` pushes, one at a time, each to a fresh + repository, with `-idle-gap` between them. This isolates the per-push cost + and shows whether memory returns after a push. +4. **Concurrency sweep.** For each K in `-conc-steps`, K pushes at once to K + fresh repositories, with an idle gap between steps. The rise per concurrent + push is the slope you size a machine with. +5. **Settle.** One more idle gap, then `heap-final.pb.gz` (forced GC) and + `allocs-final.pb.gz`. + +Every repository is created fresh under a UUID, so no push is ever measured +against a repository that already holds its objects. With the defaults that is +20 repositories. + +## Repository names + +Repositories are `{-org}/{uuidv4}`, defaulting to `benchtest/`. That is +the exact `{orgID}/{repoName}` shape `internal/repofs.Parse` requires, and it +means every benchmark repository shares one key prefix in the bucket. + +The harness never deletes anything. It writes `repos.txt` listing every +repository it created, and cleanup is yours to run. + +## Output + +Each run gets its own timestamped directory under `-out`: + +| File | What it holds | +| ------------------------ | -------------------------------------------------------------------- | +| `report.md` | The tables. Start here. | +| `samples.csv` | One row per sample, tagged with phase and repository. Plot-ready. | +| `heap-baseline.pb.gz` | Settled heap before any push. | +| `heap-peak-*.pb.gz` | Heap at each new resident-set high-water mark. | +| `heap-final.pb.gz` | Settled heap after every push. | +| `allocs-final.pb.gz` | Total allocation over the run. | +| `daemon.log` | The daemon's own JSON log. | +| `repos.txt` | Every repository created, for cleanup. | +| `source.git`, `objgitd` | The mirror and the binary under test. | + +The two commands that get the most out of a run: + +```sh +# What the pushes left behind. +go tool pprof -http=: -base /heap-baseline.pb.gz /heap-final.pb.gz + +# What was on the heap at the worst moment. +go tool pprof -http=: /heap-peak-.pb.gz +``` + +## Reading the numbers + +- **`VmHWM` is the peak**, not `VmRSS`. It is the kernel's own high-water mark, + so it cannot be missed between two samples the way a spike in `VmRSS` can. +- **Resident set is not the heap.** Go returns freed pages to the kernel lazily, + so `VmRSS` staying high after a push is not by itself a leak. The column that + settles it is `heap_inuse` in the CSV, and the baseline-diffed heap profile. + If `heap_inuse` also stays high while the daemon is idle, that is retention. +- **Peak profiles are taken with `gc=0`.** Forcing a collection would change the + number that triggered the capture. +- **Every figure is relative to `GOGC` and `GOMEMLIMIT`.** The report records + both, because a peak-RSS number without them cannot be compared to anything. +- **Wall clock includes network time** to the bucket and varies between runs. + Memory does not depend on it; throughput comparisons across runs do. + +## Flags worth knowing + +| Flag | Default | Meaning | +| ----------------------- | -------------------- | ------------------------------------------------------------------------ | +| `-repo` | `$HOME/Code/Xe/x` | Repository to push. Mirror-cloned once, never written to. | +| `-org` | `benchtest` | Org segment every benchmark repository is created under. | +| `-seq-pushes` | `5` | Sequential pushes, each to a fresh repository. | +| `-conc-steps` | `1,2,4,8` | Concurrency levels to sweep. Empty skips the sweep. | +| `-sample-interval` | `250ms` | How often `/proc` and `/metrics` are read. | +| `-idle-gap` | `5s` | Idle time between pushes, so memory can settle. | +| `-window-slack` | `0` (uses idle gap) | How far outside a push's wall clock memory is still attributed to it. | +| `-peak-growth` | `0.05` | Fractional rise in resident set that triggers a heap capture. | +| `-peak-cooldown` | `2s` | Minimum time between two peak captures. | +| `-daemon-binary` | build it | Prebuilt `objgitd` to test instead of building `./cmd/objgitd`. | +| `-daemon-allow-hooks` | `false` | Off so hook cost is not mistaken for push cost. | + +All flags take an environment fallback through `flagenv`, in UPPER_SNAKE. The +daemon-facing flags are prefixed `-daemon-` so they do not collide with +`objgitd`'s own `HTTP_BIND` and `METRICS_BIND`. + +A cheap smoke run, three repositories and well under a minute: + +```text +go run ./cmd/membench -seq-pushes 1 -conc-steps 2 -baseline 3s -idle-gap 3s +``` From 275d4ec058b7779e766c68d1148272920c47852c Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Mon, 31 Aug 2026 16:30:02 -0400 Subject: [PATCH 2/3] feat(storage/tigris): add container-writing object storer First stage of replacing the scratch go-git repository that PackfileWriter decodes a push into. Not yet wired to the write path: nothing calls segmentStore, and behaviour is unchanged. A push currently materializes objects three times over. Parser.Parse resolves every delta into scratch, planObjects resolves every delta again to read hash, type and size, and payloadFor materializes a third time through deltaForm to pick a delta form. Measured with cmd/membench, one push of a 48 MiB pack allocates about 4.4 GB, roughly 93x the pack, and planObjects alone accounts for 66% of it. Live-heap profiles understate this badly, because those objects are freed as soon as their metadata is read. segmentStore implements storer.EncodedObjectStorer over the existing packSegment staging files, so go-git's parser can write each resolved object straight into a .bin container instead of into a second git repository. RawObjectWriter stages an object to its own temp file while hashing, then hands it to packSegment.add, which keeps every container format decision in one place. Staging files stay open for the whole push because the parser reads bases back out, and a base can live in a container that filled long before the delta needing it arrives. An index miss falls through to the backing Storer, which is the thin-pack path. The plan document records why the temp .pack cannot also be removed: packfile.Parser drops to its high-memory path when the scanner has no seeker, so parsing straight from the socket would cost more memory, not less. Refs: docs/plans/streaming-container-sink.md Assisted-by: Claude Opus 5 via Claude Code Claude-Session: https://claude.ai/code/session_01PmugaKamcmXCxM4u4w29iD Signed-off-by: Xe Iaso --- docs/plans/streaming-container-sink.md | 258 +++++++++++++ internal/storage/tigris/segmentstore.go | 368 ++++++++++++++++++ internal/storage/tigris/segmentstore_test.go | 385 +++++++++++++++++++ 3 files changed, 1011 insertions(+) create mode 100644 docs/plans/streaming-container-sink.md create mode 100644 internal/storage/tigris/segmentstore.go create mode 100644 internal/storage/tigris/segmentstore_test.go diff --git a/docs/plans/streaming-container-sink.md b/docs/plans/streaming-container-sink.md new file mode 100644 index 0000000..1fd66cb --- /dev/null +++ b/docs/plans/streaming-container-sink.md @@ -0,0 +1,258 @@ +# Replace the scratch go-git storage with a container sink + +**Status:** stage 1 of 4 implemented (`internal/storage/tigris/segmentstore.go`, +9 tests, full suite green). Stages 2 to 4 not started. + +Landed separately while this was being written, both from the earlier specs: +`b130832` caps the zstd encoder concurrency, `877cd16` bounds concurrent pushes. +Those change the memory baseline this plan should be measured against, so +re-baseline with `cmd/membench` before evaluating stage 3. + +Supersedes `var/no-materializing-deltas.md`, whose subject +(`planObjects`) this deletes outright. + +## Context + +A push currently materializes objects three times over and round-trips the +whole pack through a second on-disk git repository: + +``` +wire -> Scanner(tee) -> dotgit.PackWriter -> temp .pack + -> buildIndex -> Parser.Parse + (resolve every delta, write into scratch) +packWriter.Close -> planObjects (resolve every delta AGAIN, for hash/type/size) + -> resolveDeltaBases (8 more filesystem.Storage handles over scratch) + -> payloadFor -> deltaForm -> Scanner.WriteObject + (materialize a THIRD time, to pick delta form) + -> .bin/.cue +``` + +### Live heap + +Site 1 (`Parser.Parse` -> `processDelta` -> `ensureContent`) was 67 to 79% of +live heap early in a push. Site 2 (`planObjects` -> `getMemoryObject` -> +`ApplyDelta`) was 63 to 82% of live heap at the concurrent peak. Every push +pays roughly 476 MiB of working set regardless of how warm the process is. + +### Allocation churn, which is the larger story + +Live heap understates site 2 badly, because the objects it materializes are +freed the moment their metadata has been read and so barely register in +`inuse_space`. Total allocation tells a different story. Measured over two +runs (13.07 GB across 3 pushes, 89.33 GB across 20), a single push of a 48 MiB +pack allocates: + +**~4.4 GB per push, about 93x the size of the pack.** + +By allocation site: + +| Site | Share | Per push | +| --- | --- | --- | +| `bytes.growSlice` | 47.9% | 2.09 GB | +| `plumbing.(*MemoryObject).Write` | 31.3% | 1.36 GB | +| everything else | 20.8% | ~0.9 GB | + +By objgit entry point, cumulative: + +``` +main.writePack 11323 MB 84.6% + packWriter.Close 10637 MB 79.5% + packWriter.planObjects 8880 MB 66.4% +``` + +**`planObjects` alone is 66% of all allocation per push**, roughly 2.9 GB. It +is the metadata walk that wants only hash, type, and size. + +The `MemoryObject.Write` traces also name a second contributor not visible in +the live-heap profiles: `payloadFor` -> `deltaForm` (`packwriter.go:333`) -> +`deltaFormOn` (`packwriter.go:339`) -> `Scanner.WriteObject`, which +re-materializes objects a third time while deciding whether to store each one +as a delta. + +**This inverts the earlier ranking.** Judged by live heap, site 1 looked like +the bigger term and this plan looked like it addressed the smaller half. Judged +by allocation, site 2 and the delta-form probe dominate, and this plan targets +roughly two thirds of the churn. Churn is also what the GC is working against, +which is why `GOGC=50` bought 24% of peak and 67% of settled heap for under 3% +of wall clock. + +The scratch storage exists to provide three things: delta resolution, random +access by hash, and after-the-fact delta-form lookup. Only the first is +irreducible. + +## The load-bearing constraint + +**The temp `.pack` file cannot be removed.** `parser.go:106` reads: + +```go +if p.scanner.seeker == nil { + p.lowMemoryMode = false +} +``` + +and with low memory mode off, `scanner.go:526` retains `oh.content` for every +object rather than re-inflating from disk on demand. Driving the parser +directly off the network reader would therefore *increase* memory, not reduce +it. + +So this plan keeps a seekable temp `.pack` and removes the `filesystem.Storage` +wrapped around it. What goes away is the `.idx` build, the loose/packed object +store, `planObjects`, `resolveDeltaBases`, and the eight worker storages. + +## Design + +`packSegment` is already the container this needs. `newPackSegment` +(`packwriter.go:402`) opens an `os.CreateTemp` staging file, `add` appends +payloads, `seg.recs` holds one `cueRecord` per object (hash, type, codec, +offset, stored, raw, base), and `seal` checksums and uploads. That is an +append-only blob plus a hash-to-offset side index, in the format the bucket +already stores. Nothing needs inventing, and notably a tar file would be +strictly worse: 512-byte header plus block padding per entry is over 11 MB of +overhead on this repository's 21637 objects, and would still need transcoding +to `.bin` before upload. + +The new flow: + +``` +wire -> Scanner(tee) -> temp .pack (unchanged, seekable) + -> pass 1: header-only scan of the temp .pack (delta topology) + -> pass 2: Parser.Parse(storage: segmentStore) (resolve once, stream out) + -> RawObjectWriter(typ, sz) -> packSegment -> .bin/.cue +``` + +### `segmentStore` + +A new type implementing `storer.EncodedObjectStorer`, backed by the +`packSegment` staging files and an in-memory index. + +- `RawObjectWriter(typ, sz)` returns a writer that appends through the existing + `writePayload` band logic, hashes as it streams (exactly as `stageWriter` + already does at `writer.go:33`), and on `Close` appends a `cueRecord` and + records `hash -> (segment, offset, stored, raw, codec)` in the index. Seals + and opens a new segment when the byte cap is reached. +- `EncodedObject(typ, hash)` serves the parser's REF-delta base lookups + (`parser.go:342`) from the index plus `ReadAt` on the staging file, running + `decodeBody` for the codec. +- `HasEncodedObject`, `EncodedObjectSize` read the index. +- `IterEncodedObjects` walks the index. +- `NewEncodedObject`, `SetEncodedObject` delegate to the existing `Storer` + implementations. +- `AddAlternate` returns the same unsupported error the `Storer` already does. +- `LowMemoryMode() bool` returns true, so the parser keeps its low-memory path. + +### Delta preservation + +This is the part that actually needs care, and it is why the header-only pass +exists. The parser hands out fully resolved content only, and `RawObjectWriter` +rejects delta types outright (`writer.go:35`), so a naive sink would store every +object whole and inflate the bucket. `TestPackfileWriterKeepsDeltas` and +`TestPackfileWriterDemotesDeltaAcrossSplit` encode the current contract and must +keep passing. + +Because the temp `.pack` is seekable, a header-only pass over it is cheap: seek +to each offset, parse the `ObjectHeader`, read `Type`, `OffsetReference`, and +`Reference`, and never inflate content. That yields `offset -> base offset` for +OFS deltas and `offset -> base hash` for REF deltas. + +To turn offsets into hashes, register a `packfile.Observer` via +`WithScannerObservers`. `parser.go:146` and `:150` call +`OnInflatedObjectHeader(oh.Type, oh.Size, oh.Offset)` and +`OnInflatedObjectContent(oh.Hash, oh.Offset, oh.Crc32, nil)` for every object, +where `oh.Type` and `oh.Size` are the *resolved* values and `oh.Offset` is the +pack offset. Joining on offset gives `hash -> base hash`, which is precisely +what `plannedObject.base` holds today, obtained without a single delta +application. + +Note the observer's `content` argument is always `nil` on this path, so +observers cannot carry object bytes. The bytes must come through +`RawObjectWriter`; this is why both mechanisms are needed rather than either +alone. + +### Ordering and the containment rule + +`Close` today plans the whole object set before writing, which lets `emit` +place bases before deltas and lets `inSeg` guarantee a delta's base is in the +same container. Streaming gives that up. + +Decision: **keep the two-pass shape, but make pass 1 cheap.** The header-only +pass already walks every object; have it build the same `order` slice +`planObjects` produces today, from headers alone. Pass 2 then drives the parser +and writes in that order. This preserves `emit`, `inSeg`, and the split-demotion +behaviour exactly, at the cost of the parser resolving in its own order rather +than ours. + +If that ordering mismatch proves awkward, the fallback is to accept pack order +and demote any delta whose base did not land in the same segment, which is +already the documented behaviour when a split strands a delta. + +### Base read-back across a sealed segment + +If a late REF delta references an object in a segment that already sealed and +uploaded, `EncodedObject` must still find it. Decision: **keep every staging +file until the push completes**, then delete. Disk, not memory, and it matches +the current lifetime of the scratch directory. + +## What this does not fix + +Site 1 remains. `ensureContent` (`parser.go:236`) inflates each delta into a +pooled buffer and patches it into another, whatever the sink is. It is +inherent: a git object's hash is over its resolved content, so every delta must +be reconstructed at least once. The floor is one resolved object plus its base +at a time. + +Expected recovery is site 2 in full, the delta-form probe in full, the `.idx` +build, and the eight worker storages. Against the churn measurement that is +about two thirds of the 4.4 GB a push allocates today. Against live heap it is +the larger of the two terms at the concurrent peak but a less dramatic figure. + +**Gate the result on `alloc_space`, not `inuse_space`.** Site 2's objects are +short-lived by construction, so an `inuse_space` comparison will understate +this change by a wide margin. `cmd/membench` writes `allocs-final.pb.gz` for +exactly this; divide its total by the number of pushes in the run. + +## Staging + +Each stage is independently testable and leaves the tree working. + +1. **DONE. `segmentStore` in isolation** (`internal/storage/tigris/segmentstore.go`). + The type, its index, and its `EncodedObjectStorer` methods, driven directly + by tests. The live write path still uses scratch, so nothing user-visible + changed. Notes from building it: + - `RawObjectWriter` stages each object to its own temp file while hashing, + then hands it to the existing `packSegment.add` as a `stagedObject`. The + staging hop is what keeps the object out of memory in one piece; reusing + `add` is what keeps the container format decisions in one place. + - Staging files stay open for the life of the store, so a base written into + a segment that filled early is still readable when its delta arrives. + - An index miss falls through to the backing `Storer`, which is the thin-pack + path; a miss in both surfaces as `plumbing.ErrObjectNotFound`. + - Duplicate objects within one pack are stored once. +2. **The header-only pass.** Topology extraction and `order` construction from + the temp `.pack`, tested against the same fixtures `planObjects` is tested + with, asserting identical output. +3. **Wire it up.** `packWriter.Close` drives `Parser.Parse` with `segmentStore` + and the observer instead of the scratch storage. This is the stage where the + existing pack tests are the gate. +4. **Delete the dead code.** `planObjects`, `resolveDeltaBases`, `payloadFor`, + `deltaScanWorkers`, and the scratch `filesystem.Storage`. + +## Risks + +- **Format correctness is not caught by a memory benchmark.** A wrong `raw` or + `typ` in a cue record writes a repository that pushes fine and reads corrupt + later. Stage 2 must assert byte-identical `cueRecord` output against the + current implementation before stage 3 switches anything over. +- **Thin packs.** `parser.go:342` falls back to `storage.EncodedObject` for a + REF delta base not in the pack. Against `segmentStore` that is an index miss; + it must then fall through to the real `Storer` and hence the bucket. The + current scratch storage has no alternates, so this path may not behave + identically today, and the difference needs establishing rather than assuming. +- **Failure atomicity.** A failed push today discards a scratch directory and + the bucket is untouched. Streaming into segments means partial containers can + exist. `packSegment.discard()` exists; the invariant that nothing seals until + the parse completes must be explicit. +- **SHA-256.** `writePack` threads `packfile.WithSHA256()` from the config + extension. Both new passes need the matching object ID size or they will + mis-parse every header. +- **go-git v6 is alpha.** This couples objgit to `Parser`, `Scanner`, + `ObjectHeader`, and `Observer`. All are exported, none are stable. diff --git a/internal/storage/tigris/segmentstore.go b/internal/storage/tigris/segmentstore.go new file mode 100644 index 0000000..f0c724e --- /dev/null +++ b/internal/storage/tigris/segmentstore.go @@ -0,0 +1,368 @@ +package tigris + +import ( + "bytes" + "errors" + "fmt" + "io" + "os" + "sync" + + "github.com/go-git/go-git/v6/plumbing" + "github.com/go-git/go-git/v6/plumbing/storer" +) + +// segmentStore is a storer.EncodedObjectStorer that lands objects straight into +// pack containers. +// +// It exists to replace the scratch go-git repository that PackfileWriter +// currently decodes a push into. go-git's packfile.Parser writes every object +// it resolves through RawObjectWriter (non-deltas from the scanner, resolved +// deltas from storeOrCache), so handing the parser one of these means each +// object is resolved exactly once and appended to a .bin as it arrives, instead +// of being written to a scratch repository and read back out. +// +// Staging files are deliberately kept open for the whole push rather than +// sealed as they fill: the parser reads bases back out through EncodedObject +// (packfile/parser.go's REF-delta path), and a base can live in a segment that +// filled long before the delta that needs it arrives. +var _ storer.EncodedObjectStorer = (*segmentStore)(nil) + +type segmentStore struct { + s *Storer + byteLimit int64 + + mu sync.Mutex + open *packSegment + filled []*packSegment + index map[plumbing.Hash]stagedRef +} + +// stagedRef locates one object inside one still-open staging file. +type stagedRef struct { + seg *packSegment + rec cueRecord +} + +func newSegmentStore(s *Storer) *segmentStore { + limit := s.maxPackBytes + if limit <= 0 { + limit = maxPackBytes + } + return &segmentStore{ + s: s, + byteLimit: limit, + index: map[plumbing.Hash]stagedRef{}, + } +} + +// LowMemoryMode reports that the parser may drop object contents and re-inflate +// them from the pack on demand. Answering true here is what keeps +// packfile.Parser off the path that retains every object's bytes in memory; +// see the seeker check in packfile/parser.go. +func (ss *segmentStore) LowMemoryMode() bool { return true } + +// segments returns every staging file written so far, the open one last. The +// caller seals and uploads them; this type never touches the bucket. +func (ss *segmentStore) segments() []*packSegment { + ss.mu.Lock() + defer ss.mu.Unlock() + + out := make([]*packSegment, 0, len(ss.filled)+1) + out = append(out, ss.filled...) + if ss.open != nil && len(ss.open.recs) > 0 { + out = append(out, ss.open) + } + return out +} + +// discard throws away every staging file. Idempotent, and safe to defer for the +// error path: sealing removes segments from this store's ownership first. +func (ss *segmentStore) discard() { + ss.mu.Lock() + defer ss.mu.Unlock() + + for _, seg := range ss.filled { + seg.discard() + } + if ss.open != nil { + ss.open.discard() + } + ss.filled, ss.open = nil, nil + ss.index = map[plumbing.Hash]stagedRef{} +} + +// segmentFor returns a segment with room for an object of size sz, rotating the +// open one out when it would overflow the byte cap. +// +// The cap is checked before the add and only when the segment already holds +// something, which is the same rule packWriter.Close applies: an object larger +// than the whole cap gets a container to itself, because it has to live +// somewhere. +func (ss *segmentStore) segmentFor(sz int64) (*packSegment, error) { + if ss.open != nil && len(ss.open.recs) > 0 && ss.open.offset+sz > ss.byteLimit { + ss.filled = append(ss.filled, ss.open) + ss.open = nil + } + if ss.open == nil { + seg, err := newPackSegment(ss.s) + if err != nil { + return nil, err + } + ss.open = seg + } + return ss.open, nil +} + +// RawObjectWriter stages one object to its own temp file, hashing as it goes, +// and appends it to a container on Close. The staging hop buys streaming: the +// object never exists in memory in one piece, which is the whole point of +// taking the parser's output this way rather than through SetEncodedObject. +func (ss *segmentStore) RawObjectWriter(typ plumbing.ObjectType, sz int64) (io.WriteCloser, error) { + if typ == plumbing.OFSDeltaObject || typ == plumbing.REFDeltaObject { + return nil, plumbing.ErrInvalidType + } + if sz < 0 { + return nil, fmt.Errorf("tigris: negative object size %d", sz) + } + + f, err := os.CreateTemp("", "objgit-tigris-seg-*") + if err != nil { + return nil, fmt.Errorf("tigris: create staging file: %w", err) + } + + return &segmentWriter{ + ss: ss, + f: f, + typ: typ, + size: sz, + hasher: plumbing.NewHasher(ss.s.of, typ, sz), + }, nil +} + +type segmentWriter struct { + ss *segmentStore + f *os.File + typ plumbing.ObjectType + size int64 + wrote int64 + hasher plumbing.Hasher + done bool +} + +func (w *segmentWriter) Write(p []byte) (int, error) { + if w.done { + return 0, errors.New("tigris: write on discarded segment writer") + } + + n, err := io.MultiWriter(w.f, w.hasher).Write(p) + w.wrote += int64(n) + if err != nil { + // File and hasher may disagree from here on; the stream is poison. + w.discard() + return n, fmt.Errorf("tigris: stage write: %w", err) + } + if w.wrote > w.size { + w.discard() + return n, fmt.Errorf("tigris: wrote %d bytes beyond declared size %d", w.wrote-w.size, w.size) + } + return n, nil +} + +func (w *segmentWriter) discard() { + if w.done { + return + } + w.done = true + w.f.Close() + os.Remove(w.f.Name()) +} + +func (w *segmentWriter) Close() error { + if w.done { + return nil + } + w.done = true + defer os.Remove(w.f.Name()) + + if w.wrote != w.size { + w.f.Close() + return fmt.Errorf("tigris: staged %d bytes but declared %d", w.wrote, w.size) + } + if err := w.f.Close(); err != nil { + return fmt.Errorf("tigris: close staging file: %w", err) + } + + h := w.hasher.Sum() + obj := &stagedObject{path: w.f.Name(), typ: w.typ, size: w.size, hash: h} + + w.ss.mu.Lock() + defer w.ss.mu.Unlock() + + // An object already staged is not written twice. A pack may legitimately + // carry the same object more than once, and the second copy would only add + // bytes to the container under a hash the index already resolves. + if _, ok := w.ss.index[h]; ok { + return nil + } + + seg, err := w.ss.segmentFor(w.size) + if err != nil { + return err + } + + before := len(seg.recs) + if err := seg.add(storedObject{payload: obj, hash: h, typ: w.typ, raw: w.size}); err != nil { + return err + } + if len(seg.recs) != before+1 { + return fmt.Errorf("tigris: segment recorded %d records for one object", len(seg.recs)-before) + } + + w.ss.index[h] = stagedRef{seg: seg, rec: seg.recs[before]} + return nil +} + +// stagedObject is one object held in its own temp file, presented to +// packSegment.add as an EncodedObject. Reader opens the file fresh each call so +// writePayload can stream it, and rewind it for the probe path, without holding +// the bytes. +type stagedObject struct { + path string + typ plumbing.ObjectType + size int64 + hash plumbing.Hash +} + +func (o *stagedObject) Hash() plumbing.Hash { return o.hash } +func (o *stagedObject) Type() plumbing.ObjectType { return o.typ } +func (o *stagedObject) SetType(t plumbing.ObjectType) { o.typ = t } +func (o *stagedObject) Size() int64 { return o.size } +func (o *stagedObject) SetSize(n int64) { o.size = n } +func (o *stagedObject) Reader() (io.ReadCloser, error) { return os.Open(o.path) } + +func (o *stagedObject) Writer() (io.WriteCloser, error) { + return nil, errors.New("tigris: staged object is read-only") +} + +// EncodedObject serves the parser's base lookups out of the staging files. A +// miss falls through to the backing Storer, which is what lets a thin pack +// resolve a delta against an object the repository already holds. +func (ss *segmentStore) EncodedObject(t plumbing.ObjectType, h plumbing.Hash) (plumbing.EncodedObject, error) { + ss.mu.Lock() + ref, ok := ss.index[h] + ss.mu.Unlock() + + if !ok { + return ss.s.EncodedObject(t, h) + } + if t != plumbing.AnyObject && ref.rec.typ != t { + return nil, plumbing.ErrObjectNotFound + } + + stored := make([]byte, ref.rec.stored) + if _, err := ref.seg.file.ReadAt(stored, ref.rec.offset); err != nil { + return nil, fmt.Errorf("tigris: read %s from staging: %w", h.String(), err) + } + + plain, err := ss.s.decodePackedPayload(h, packEntry{ + typ: ref.rec.typ, + codec: ref.rec.codec, + offset: ref.rec.offset, + stored: ref.rec.stored, + raw: ref.rec.raw, + }, bytes.NewReader(stored)) + if err != nil { + return nil, err + } + + obj := plumbing.NewMemoryObject(ss.s.oh) + obj.SetType(ref.rec.typ) + obj.SetSize(ref.rec.raw) + if _, err := obj.Write(plain); err != nil { + return nil, fmt.Errorf("tigris: rebuild %s: %w", h.String(), err) + } + return obj, nil +} + +func (ss *segmentStore) HasEncodedObject(h plumbing.Hash) error { + ss.mu.Lock() + _, ok := ss.index[h] + ss.mu.Unlock() + + if ok { + return nil + } + return ss.s.HasEncodedObject(h) +} + +func (ss *segmentStore) EncodedObjectSize(h plumbing.Hash) (int64, error) { + ss.mu.Lock() + ref, ok := ss.index[h] + ss.mu.Unlock() + + if ok { + return ref.rec.raw, nil + } + return ss.s.EncodedObjectSize(h) +} + +// NewEncodedObject and SetEncodedObject delegate: the parser reaches for +// RawObjectWriter, so these exist to satisfy the interface and to keep any +// caller that does use them behaving exactly as the Storer does. +func (ss *segmentStore) NewEncodedObject() plumbing.EncodedObject { return ss.s.NewEncodedObject() } + +func (ss *segmentStore) SetEncodedObject(obj plumbing.EncodedObject) (plumbing.Hash, error) { + return ss.s.SetEncodedObject(obj) +} + +func (ss *segmentStore) IterEncodedObjects(t plumbing.ObjectType) (storer.EncodedObjectIter, error) { + ss.mu.Lock() + hashes := make([]plumbing.Hash, 0, len(ss.index)) + for h, ref := range ss.index { + if t == plumbing.AnyObject || ref.rec.typ == t { + hashes = append(hashes, h) + } + } + ss.mu.Unlock() + + return &segmentIter{ss: ss, typ: t, hashes: hashes}, nil +} + +func (ss *segmentStore) AddAlternate(remote string) error { return ss.s.AddAlternate(remote) } + +type segmentIter struct { + ss *segmentStore + typ plumbing.ObjectType + hashes []plumbing.Hash + i int +} + +func (it *segmentIter) Next() (plumbing.EncodedObject, error) { + if it.i >= len(it.hashes) { + return nil, io.EOF + } + h := it.hashes[it.i] + it.i++ + return it.ss.EncodedObject(it.typ, h) +} + +func (it *segmentIter) ForEach(fn func(plumbing.EncodedObject) error) error { + for { + obj, err := it.Next() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if err := fn(obj); err != nil { + if errors.Is(err, storer.ErrStop) { + return nil + } + return err + } + } +} + +func (it *segmentIter) Close() { it.i = len(it.hashes) } diff --git a/internal/storage/tigris/segmentstore_test.go b/internal/storage/tigris/segmentstore_test.go new file mode 100644 index 0000000..0a06c9c --- /dev/null +++ b/internal/storage/tigris/segmentstore_test.go @@ -0,0 +1,385 @@ +package tigris + +import ( + "bytes" + "errors" + "fmt" + "io" + "os" + "testing" + + "github.com/go-git/go-git/v6/plumbing" +) + +// putRaw writes one object through the segment store the way packfile.Parser +// does, and returns the hash it landed under. +func putRaw(t *testing.T, ss *segmentStore, typ plumbing.ObjectType, body []byte) plumbing.Hash { + t.Helper() + + w, err := ss.RawObjectWriter(typ, int64(len(body))) + if err != nil { + t.Fatalf("RawObjectWriter: %v", err) + } + if _, err := w.Write(body); err != nil { + t.Fatalf("write: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + h := plumbing.NewHasher(ss.s.of, typ, int64(len(body))) + if _, err := h.Write(body); err != nil { + t.Fatalf("hash: %v", err) + } + return h.Sum() +} + +func readBack(t *testing.T, ss *segmentStore, h plumbing.Hash) (plumbing.ObjectType, []byte) { + t.Helper() + + obj, err := ss.EncodedObject(plumbing.AnyObject, h) + if err != nil { + t.Fatalf("EncodedObject(%s): %v", h, err) + } + rd, err := obj.Reader() + if err != nil { + t.Fatalf("reader: %v", err) + } + defer rd.Close() + + got, err := io.ReadAll(rd) + if err != nil { + t.Fatalf("read: %v", err) + } + return obj.Type(), got +} + +func TestSegmentStoreRoundTrip(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + typ plumbing.ObjectType + body []byte + }{ + {name: "tiny blob below the compression floor", typ: plumbing.BlobObject, body: []byte("hello")}, + {name: "empty blob", typ: plumbing.BlobObject, body: []byte{}}, + {name: "commit", typ: plumbing.CommitObject, body: []byte("tree deadbeef\n\nmessage\n")}, + {name: "compressible blob above the floor", typ: plumbing.BlobObject, body: bytes.Repeat([]byte("abcd"), 4096)}, + {name: "incompressible blob above the floor", typ: plumbing.BlobObject, body: pseudoRandom(8192)}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + h := putRaw(t, ss, tt.typ, tt.body) + + gotTyp, gotBody := readBack(t, ss, h) + if gotTyp != tt.typ { + t.Logf("want: %v", tt.typ) + t.Logf("got: %v", gotTyp) + t.Error("wrong type") + } + if !bytes.Equal(gotBody, tt.body) { + t.Logf("want %d bytes", len(tt.body)) + t.Logf("got %d bytes", len(gotBody)) + t.Error("payload did not round-trip") + } + + size, err := ss.EncodedObjectSize(h) + if err != nil { + t.Fatalf("EncodedObjectSize: %v", err) + } + if size != int64(len(tt.body)) { + t.Logf("want: %d", len(tt.body)) + t.Logf("got: %d", size) + t.Error("wrong recorded size") + } + if err := ss.HasEncodedObject(h); err != nil { + t.Errorf("HasEncodedObject: %v", err) + } + }) + } +} + +func TestSegmentStoreRejections(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + typ plumbing.ObjectType + size int64 + err error + }{ + {name: "ref delta", typ: plumbing.REFDeltaObject, size: 4, err: plumbing.ErrInvalidType}, + {name: "ofs delta", typ: plumbing.OFSDeltaObject, size: 4, err: plumbing.ErrInvalidType}, + {name: "negative size", typ: plumbing.BlobObject, size: -1}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + _, err := ss.RawObjectWriter(tt.typ, tt.size) + if err == nil { + t.Fatal("wanted an error, got none") + } + if tt.err != nil && !errors.Is(err, tt.err) { + t.Logf("want: %v", tt.err) + t.Logf("got: %v", err) + t.Error("wrong error") + } + }) + } +} + +func TestSegmentStoreShortAndOverlongWrites(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + declare int64 + body []byte + failsOn string + }{ + {name: "fewer bytes than declared", declare: 10, body: []byte("abc"), failsOn: "close"}, + {name: "more bytes than declared", declare: 3, body: []byte("abcdefgh"), failsOn: "write"}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + w, err := ss.RawObjectWriter(plumbing.BlobObject, tt.declare) + if err != nil { + t.Fatalf("RawObjectWriter: %v", err) + } + + _, writeErr := w.Write(tt.body) + closeErr := w.Close() + + switch tt.failsOn { + case "write": + if writeErr == nil { + t.Error("wanted the write to fail, got no error") + } + case "close": + if writeErr != nil { + t.Fatalf("unexpected write error: %v", writeErr) + } + if closeErr == nil { + t.Error("wanted the close to fail, got no error") + } + } + + if got := len(ss.segments()); got != 0 { + t.Logf("want: 0 segments") + t.Logf("got: %d", got) + t.Error("a rejected object still opened a container") + } + }) + } +} + +func TestSegmentStoreSplitsAtByteLimit(t *testing.T) { + t.Parallel() + + // Four objects of 1 KiB against a 2 KiB cap: the third has to open a second + // container, and the cap is checked before the add, so no container ever + // exceeds it. + ss := newSegmentStore(newTestStorer(t, newFakeS3(t), withMaxPackBytes(2<<10))) + defer ss.discard() + + var hashes []plumbing.Hash + for i := range 4 { + hashes = append(hashes, putRaw(t, ss, plumbing.BlobObject, pseudoRandomSeeded(1<<10, i))) + } + + segs := ss.segments() + if len(segs) < 2 { + t.Logf("want: at least 2 segments") + t.Logf("got: %d", len(segs)) + t.Fatal("byte cap did not split the container") + } + + total := 0 + for _, seg := range segs { + total += len(seg.recs) + if seg.offset > 2<<10 && len(seg.recs) > 1 { + t.Errorf("segment holds %d bytes over a %d cap with %d records", seg.offset, 2<<10, len(seg.recs)) + } + } + if total != 4 { + t.Logf("want: 4 records across all segments") + t.Logf("got: %d", total) + t.Error("objects went missing across the split") + } + + // Every object stays readable after its container filled, which is what + // lets the parser resolve a base written long before its delta arrives. + for i, h := range hashes { + if _, body := readBack(t, ss, h); len(body) != 1<<10 { + t.Errorf("object %d read back %d bytes, want %d", i, len(body), 1<<10) + } + } +} + +func TestSegmentStoreDeduplicates(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + body := []byte("the same object twice") + first := putRaw(t, ss, plumbing.BlobObject, body) + second := putRaw(t, ss, plumbing.BlobObject, body) + + if first != second { + t.Fatal("the same bytes hashed differently") + } + + segs := ss.segments() + if len(segs) != 1 { + t.Fatalf("want 1 segment, got %d", len(segs)) + } + if got := len(segs[0].recs); got != 1 { + t.Logf("want: 1 record") + t.Logf("got: %d", got) + t.Error("a duplicate object was stored twice") + } +} + +func TestSegmentStoreMissFallsThroughToStorer(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + // Nothing staged, so this can only be answered by the backing Storer. The + // object is not there either, so the fall-through must surface as a + // not-found rather than as a staging error. + var absent plumbing.Hash + absent = plumbing.NewHasher(ss.s.of, plumbing.BlobObject, 3).Sum() + + _, err := ss.EncodedObject(plumbing.AnyObject, absent) + if err == nil { + t.Fatal("wanted an error for an object in neither place, got none") + } + if !errors.Is(err, plumbing.ErrObjectNotFound) { + t.Logf("want: %v", plumbing.ErrObjectNotFound) + t.Logf("got: %v", err) + t.Error("wrong error from the fall-through path") + } +} + +func TestSegmentStoreTypeFilter(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + h := putRaw(t, ss, plumbing.BlobObject, []byte("a blob")) + + if _, err := ss.EncodedObject(plumbing.BlobObject, h); err != nil { + t.Errorf("matching type should resolve: %v", err) + } + _, err := ss.EncodedObject(plumbing.CommitObject, h) + if !errors.Is(err, plumbing.ErrObjectNotFound) { + t.Logf("want: %v", plumbing.ErrObjectNotFound) + t.Logf("got: %v", err) + t.Error("a type mismatch should read as not found") + } +} + +func TestSegmentStoreIter(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t))) + defer ss.discard() + + want := map[plumbing.Hash]bool{} + for i := range 3 { + want[putRaw(t, ss, plumbing.BlobObject, []byte(fmt.Sprintf("blob %d", i)))] = true + } + commit := putRaw(t, ss, plumbing.CommitObject, []byte("a commit")) + + iter, err := ss.IterEncodedObjects(plumbing.BlobObject) + if err != nil { + t.Fatalf("IterEncodedObjects: %v", err) + } + got := map[plumbing.Hash]bool{} + if err := iter.ForEach(func(o plumbing.EncodedObject) error { + got[o.Hash()] = true + return nil + }); err != nil { + t.Fatalf("ForEach: %v", err) + } + + if len(got) != len(want) { + t.Logf("want: %d blobs", len(want)) + t.Logf("got: %d", len(got)) + t.Error("wrong number of blobs") + } + for h := range want { + if !got[h] { + t.Errorf("blob %s missing from the iteration", h) + } + } + if got[commit] { + t.Error("the commit leaked into a blob-only iteration") + } +} + +func TestSegmentStoreDiscardLeavesNothing(t *testing.T) { + t.Parallel() + + ss := newSegmentStore(newTestStorer(t, newFakeS3(t), withMaxPackBytes(2<<10))) + for i := range 4 { + putRaw(t, ss, plumbing.BlobObject, pseudoRandomSeeded(1<<10, i)) + } + + paths := []string{} + for _, seg := range ss.segments() { + paths = append(paths, seg.path) + } + if len(paths) < 2 { + t.Fatalf("want at least 2 staging files, got %d", len(paths)) + } + + ss.discard() + + for _, p := range paths { + if fileExists(p) { + t.Errorf("staging file %s survived discard", p) + } + } + if got := len(ss.segments()); got != 0 { + t.Errorf("want 0 segments after discard, got %d", got) + } +} + +// pseudoRandom returns n bytes that zstd cannot shrink, so a test can reach the +// raw-codec branch deliberately. +func pseudoRandom(n int) []byte { return pseudoRandomSeeded(n, 0) } + +// fileExists reports whether a staging path is still on disk. +func fileExists(path string) bool { + _, err := os.Stat(path) + return err == nil +} + +func pseudoRandomSeeded(n, seed int) []byte { + out := make([]byte, n) + x := uint64(seed)*2862933555777941757 + 3037000493 + for i := range out { + x ^= x << 13 + x ^= x >> 7 + x ^= x << 17 + out[i] = byte(x) + } + return out +} From 00718b7a58cb122ffac9e5b5c409fbcba729c0ee Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Mon, 31 Aug 2026 16:32:30 -0400 Subject: [PATCH 3/3] style(storage/tigris): merge declaration with assignment staticcheck S1021. The split declaration was left over from an earlier version of the test that assigned the hash conditionally. Assisted-by: Claude Opus 5 via Claude Code Claude-Session: https://claude.ai/code/session_01PmugaKamcmXCxM4u4w29iD Signed-off-by: Xe Iaso --- internal/storage/tigris/segmentstore_test.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/internal/storage/tigris/segmentstore_test.go b/internal/storage/tigris/segmentstore_test.go index 0a06c9c..61f8141 100644 --- a/internal/storage/tigris/segmentstore_test.go +++ b/internal/storage/tigris/segmentstore_test.go @@ -262,8 +262,7 @@ func TestSegmentStoreMissFallsThroughToStorer(t *testing.T) { // Nothing staged, so this can only be answered by the backing Storer. The // object is not there either, so the fall-through must surface as a // not-found rather than as a staging error. - var absent plumbing.Hash - absent = plumbing.NewHasher(ss.s.of, plumbing.BlobObject, 3).Sum() + absent := plumbing.NewHasher(ss.s.of, plumbing.BlobObject, 3).Sum() _, err := ss.EncodedObject(plumbing.AnyObject, absent) if err == nil {