feat: erofs snapshots of pushed trees, stored next to the repo

AGENTS.md

@@ -59,13 +59,15 @@ Notes on the tests and on configuration:
 | `cmd/objgitd/shell.go`                                    | The SSH `sh` command: an interactive shell in the hook sandbox.                   |
 | `cmd/objgitd/receivepack.go`                              | The go-git fork that streams hook output, plus `writePack`.                       |
 | `cmd/objgitd/hooks.go`                                    | Ref diffing and the sandboxed hook run.                                           |
+| `cmd/objgitd/snapshots.go`                                | The erofs snapshot run after a push.                                              |
 | `internal/auth`                                           | The one authorization interface.                                                  |
 | `internal/repofs`                                         | Maps a repository path to a `storage.Storer`.                                     |
 | `internal/storage/tigris`                                 | Repository storage. A `storage.Storer` on the bucket.                             |
 | `internal/bundler`                                        | The async upload queue behind that storer.                                        |
 | `internal/s3fs`                                           | Daemon-level state only, which is the SSH host key.                               |
 | `internal/mountfs`, `internal/treefs`, `internal/kefkash` | The hook sandbox filesystem and shell wiring.                                     |
 | `internal/metrics`                                        | Every Prometheus vector, plus thin helpers.                                       |
+| `internal/snapshot`                                       | erofs images of git trees: `Ensure`, `Open`, and the `Store` interface.           |
 | `internal/slog.go`                                        | JSON handler init.                                                                |
 | `cmd/membench/`                                           | Push memory benchmark harness. Not shipped; see `docs/usage/memory-benchmark.md`. |
 
@@ -80,6 +82,7 @@ describes the daemon and links to one page for each subsystem.
 | [auth.md](docs/architecture/auth.md)                   | Credentials, decisions, or a new `Authorizer`.                          |
 | [hooks.md](docs/architecture/hooks.md)                 | Push hooks, output streaming, or the sandbox.                           |
 | [metrics.md](docs/architecture/metrics.md)             | Any metric or instrumentation seam.                                     |
+| [snapshots.md](docs/architecture/snapshots.md)         | Snapshot images, the snapshot cache, or `runSnapshots`.                 |
 | [tigris-storer.md](docs/architecture/tigris-storer.md) | Object layout, refs, packs, the pack cache, or the upload path.         |
 | [s3fs.md](docs/architecture/s3fs.md)                   | The `billy.Filesystem` over the bucket.                                 |
 

cmd/membench/main.go

@@ -62,6 +62,7 @@ var (
 	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")
+	daemonSnapshots  = flag.Bool("daemon-erofs-snapshots", false, "build erofs snapshots in the daemon under test; off by default so snapshot 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)")
@@ -137,6 +138,7 @@ func run(ctx context.Context) error {
 		"-ssh-bind=",
 		"-allow-push",
 		fmt.Sprintf("-allow-hooks=%t", *daemonAllowHooks),
+		fmt.Sprintf("-erofs-snapshots=%t", *daemonSnapshots),
 		"-pack-cache-dir", packCache,
 	}
 

cmd/objgitd/git_protocol.go

@@ -54,6 +54,12 @@ type daemon struct {
 	allowHooks  bool
 	hookTimeout time.Duration
 
+	// snapshots gates building an erofs image of each updated ref tip after
+	// a push; see snapshots.go.
+	snapshots       bool
+	snapshotTimeout time.Duration
+	snapshotTmpDir  string // parent directory for image temp files; "" is the OS temp dir
+
 	// pushes bounds how many pushes unpack a packfile at once, which is the
 	// only thing that bounds the daemon's resident set under concurrent pushes.
 	// A nil pushes is unlimited, which is what every test that does not care

cmd/objgitd/hooks.go

@@ -45,8 +45,9 @@ type refUpdate struct {
 }
 
 // receivePack runs the receive-pack service and dispatches post-receive work
-// for the commands whose ref updates succeeded. Hook output streams to the
-// client before the response closes. Webhook delivery also finishes before the
+// for the commands whose ref updates succeeded: erofs snapshots of the updated
+// branch and tag tips (see runSnapshots), then hooks, then webhooks. Snapshot
+// and hook output streams to the client before the response closes. Webhook delivery also finishes before the
 // response closes, but cannot reject an already accepted push.
 //
 // This is also the one place every push funnels through — smart HTTP and SSH
@@ -57,7 +58,7 @@ func (d *daemon) receivePack(ctx context.Context, st storage.Storer, repoPath st
 	if settingsErr != nil {
 		slog.Error("webhook: load repository settings", "repo", repoPath, "err", settingsErr)
 	}
-	if !d.allowHooks && (webhooks == nil || !webhooks.Enabled(repoPath)) {
+	if !d.allowHooks && !d.snapshots && (webhooks == nil || !webhooks.Enabled(repoPath)) {
 		err := receivePackStreaming(ctx, st, r, w, req, d.pushes.admit, nil)
 		d.healHEADAfterPush(err, st, repoPath)
 		return err
@@ -67,6 +68,9 @@ func (d *daemon) receivePack(ctx context.Context, st storage.Storer, repoPath st
 	// updates are taken from this request's successful commands, so another
 	// concurrent push cannot be attributed to this one.
 	onUpdated := func(progress io.Writer, updates []refUpdate, acceptedAt time.Time) {
+		if d.snapshots {
+			d.runSnapshots(repoPath, st, updates, progress)
+		}
 		if d.allowHooks {
 			var branches []refUpdate
 			for _, u := range updates {
@@ -102,8 +106,8 @@ func (d *daemon) healHEADAfterPush(recvErr error, st storage.Storer, repoPath st
 // streaming each hook's output to progress (nil = log only).
 func (d *daemon) runHooks(repoPath, service string, st storage.Storer, updates []refUpdate, progress io.Writer) {
 	for _, u := range updates {
-		if u.New.IsZero() {
-			continue // branch deletion: nothing to check out
+		if u.New.IsZero() || !u.Name.IsBranch() {
+			continue // a deletion, or a tag: hooks run for branches only
 		}
 		d.runHook(repoPath, service, st, u, progress)
 	}

cmd/objgitd/main.go

@@ -46,6 +46,11 @@ var (
 	packCacheDir   = flag.String("pack-cache-dir", "", "parent directory for the local pack cache; empty uses the OS temp directory")
 	packCacheBytes = flag.Int64("pack-cache-bytes", 2<<30, "disk budget for the local pack cache, least-recently-used eviction; 0 disables caching")
 
+	erofsSnapshots       = flag.Bool("erofs-snapshots", true, "build a zstd-compressed erofs image of the tree at each updated branch and tag tip after a push, and store it next to the repository; the push waits for the build")
+	snapshotTimeout      = flag.Duration("snapshot-timeout", 2*time.Minute, "wall-clock limit for the erofs snapshots of one push")
+	snapshotCacheBytes   = flag.Int64("snapshot-cache-bytes", 2<<30, "disk budget for the local erofs snapshot cache, least-recently-used eviction; 0 disables caching")
+	snapshotCacheMaxIdle = flag.Duration("snapshot-cache-max-idle", time.Hour, "evict a cached erofs snapshot that nobody opened for this long; 0 disables the idle sweep")
+
 	packCompression = flag.Bool("pack-compression", true, "store zstd-compressed payloads in newly written pack containers; reading compressed containers is always enabled, so this is safe to turn off for one release before a rollback")
 	packedRefs      = flag.Bool("packed-refs", true, "write every ref into one packed-refs object under a compare-and-swap, instead of one object per ref; reading packed refs is always enabled, so this is safe to turn off for one release before a rollback")
 
@@ -131,6 +136,18 @@ func main() {
 		storerOpts = append(storerOpts, tigris.WithPackCache(packCache))
 	}
 
+	// The snapshot cache sits next to the pack cache, under the same parent,
+	// with its own budget. snapshot.Open downloads whole images into it.
+	var snapCache *tigris.PackCache
+	if *snapshotCacheBytes > 0 {
+		snapCache, err = tigris.NewSnapshotCache(*packCacheDir, *snapshotCacheBytes, metrics.ObserveSnapshotCache)
+		if err != nil {
+			slog.Error("can't create snapshot cache", "pack_cache_dir", *packCacheDir, "err", err)
+			os.Exit(1)
+		}
+		storerOpts = append(storerOpts, tigris.WithSnapshotCache(snapCache))
+	}
+
 	// Every repository lives in the same bucket as daemon system state, keyed
 	// by an "orgID/name" prefix (repofs.BucketResolver via tigrisBase.Scoped).
 	base, err := tigris.New(ctx, *bucket, storerOpts...)
@@ -146,6 +163,10 @@ func main() {
 		allowHooks:  *allowHooks,
 		hookTimeout: *hookTimeout,
 		pushes:      newPushLimiter(*maxConcurrentPushes, *pushQueueTimeout),
+
+		snapshots:       *erofsSnapshots,
+		snapshotTimeout: *snapshotTimeout,
+		snapshotTmpDir:  *packCacheDir,
 	}
 
 	slog.Info("objgitd listening",
@@ -157,12 +178,33 @@ func main() {
 		"allow_push", *allowPush,
 		"allow_hooks", *allowHooks,
 		"pack_cache_bytes", *packCacheBytes,
+		"erofs_snapshots", *erofsSnapshots,
+		"snapshot_cache_bytes", *snapshotCacheBytes,
 		"max_concurrent_pushes", *maxConcurrentPushes,
 		"push_queue_timeout", *pushQueueTimeout,
 	)
 
 	g, gCtx := errgroup.WithContext(ctx)
 
+	// The budget alone only evicts when a new image arrives. This sweep also
+	// cycles out images that nobody opened for a while.
+	if snapCache != nil && *snapshotCacheMaxIdle > 0 {
+		g.Go(func() error {
+			tick := time.NewTicker(*snapshotCacheMaxIdle / 4)
+			defer tick.Stop()
+			for {
+				select {
+				case <-gCtx.Done():
+					return nil
+				case <-tick.C:
+					if n := snapCache.EvictIdle(*snapshotCacheMaxIdle); n > 0 {
+						slog.Debug("evicted idle snapshots", "count", n)
+					}
+				}
+			}
+		})
+	}
+
 	if *metricsBind != "" {
 		ln, err := net.Listen("tcp", *metricsBind)
 		if err != nil {
@@ -233,6 +275,9 @@ func main() {
 	if cerr := packCache.Cleanup(); cerr != nil {
 		slog.Warn("can't remove the pack cache directory", "err", cerr)
 	}
+	if cerr := snapCache.Cleanup(); cerr != nil {
+		slog.Warn("can't remove the snapshot cache directory", "err", cerr)
+	}
 
 	if err != nil {
 		slog.Error("server stopped", "err", err)

cmd/objgitd/snapshots.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/go-git/go-git/v6/plumbing"
+	"github.com/go-git/go-git/v6/plumbing/object"
+	"github.com/go-git/go-git/v6/plumbing/storer"
+	"github.com/go-git/go-git/v6/storage"
+	"github.com/tigrisdata/objgit/internal/metrics"
+	"github.com/tigrisdata/objgit/internal/snapshot"
+)
+
+// maxTagDepth bounds how many annotated tags peelToTree follows.
+const maxTagDepth = 16
+
+// runSnapshots makes sure that an erofs image exists for the tree at the tip
+// of each updated branch and tag. It runs synchronously inside onUpdated, so
+// the push waits for it, and it writes one progress line per ref. A failure is logged
+// and counted and never fails the push: the refs are already committed.
+func (d *daemon) runSnapshots(repoPath string, st storage.Storer, updates []refUpdate, progress io.Writer) {
+	store, ok := st.(snapshot.Store)
+	if !ok {
+		slog.Debug("snapshot: storer holds no snapshot store, skipping", "repo", repoPath)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), d.snapshotTimeout)
+	defer cancel()
+
+	// Branches go first, then tags, each by name, so the ref that builds a
+	// shared tree (and the progress output) is the same on every push, however
+	// the client ordered its commands.
+	ordered := slices.Clone(updates)
+	slices.SortFunc(ordered, func(a, b refUpdate) int {
+		if a.Name.IsBranch() != b.Name.IsBranch() {
+			if a.Name.IsBranch() {
+				return -1
+			}
+			return 1
+		}
+		return strings.Compare(a.Name.String(), b.Name.String())
+	})
+
+	// A branch and a tag on one commit share one tree, and one build.
+	seen := map[plumbing.Hash]string{}
+	for _, u := range ordered {
+		if u.New.IsZero() || !(u.Name.IsBranch() || u.Name.IsTag()) {
+			continue // a deletion, or a ref such as refs/notes/*
+		}
+		log := slog.With("repo", repoPath, "ref", u.Name.String(), "sha", u.New.String())
+
+		tree, ok, err := peelToTree(st, u.New)
+		if err != nil {
+			log.Error("snapshot: resolve tree", "err", err)
+			metrics.ObserveSnapshot("error", 0, 0)
+			continue
+		}
+		if !ok {
+			log.Debug("snapshot: ref does not point at a commit, skipping")
+			continue
+		}
+		if status, dup := seen[tree]; dup {
+			if status == snapshot.StatusBuilt {
+				status = snapshot.StatusExists
+			}
+			writeSnapshotLine(progress, u.Name, tree, status)
+			continue
+		}
+
+		res, err := snapshot.Ensure(ctx, st, store, tree, d.snapshotTmpDir)
+		if err != nil {
+			log.Error("snapshot: build failed", "tree", tree.String(), "dur", res.Elapsed, "err", err)
+			metrics.ObserveSnapshot("error", res.Elapsed, 0)
+			seen[tree] = "failed"
+			writeSnapshotLine(progress, u.Name, tree, "failed")
+			continue
+		}
+		metrics.ObserveSnapshot(res.Status, res.Elapsed, res.Bytes)
+		seen[tree] = res.Status
+		if res.Status == snapshot.StatusBuilt {
+			log.Info("snapshot: built", "key", res.Key, "files", res.Files, "bytes", res.Bytes, "dur", res.Elapsed)
+			writeSnapshotLine(progress, u.Name, tree, fmt.Sprintf("built, %d files, %s, %s",
+				res.Files, humanBytes(res.Bytes), res.Elapsed.Round(100*time.Millisecond)))
+			continue
+		}
+		writeSnapshotLine(progress, u.Name, tree, res.Status)
+	}
+}
+
+// writeSnapshotLine writes one "remote: objgit: snapshot ..." line. It never
+// carries error text, which can name bucket internals; the log has that.
+func writeSnapshotLine(progress io.Writer, ref plumbing.ReferenceName, tree plumbing.Hash, status string) {
+	if progress == nil {
+		return
+	}
+	fmt.Fprintf(progress, "objgit: snapshot %s (tree %s): %s\n", ref, tree.String()[:7], status)
+}
+
+// peelToTree returns the tree of the commit that h names, following annotated
+// tags. ok is false, with no error, when h does not lead to a commit (a tag of
+// a tree or a blob, or a tree itself), since there is then nothing to snapshot.
+func peelToTree(st storer.EncodedObjectStorer, h plumbing.Hash) (plumbing.Hash, bool, error) {
+	for range maxTagDepth {
+		obj, err := st.EncodedObject(plumbing.AnyObject, h)
+		if err != nil {
+			return plumbing.ZeroHash, false, fmt.Errorf("load %s: %w", h, err)
+		}
+		switch obj.Type() {
+		case plumbing.CommitObject:
+			c, err := object.DecodeCommit(st, obj)
+			if err != nil {
+				return plumbing.ZeroHash, false, fmt.Errorf("decode commit %s: %w", h, err)
+			}
+			return c.TreeHash, true, nil
+		case plumbing.TagObject:
+			tag, err := object.DecodeTag(st, obj)
+			if err != nil {
+				return plumbing.ZeroHash, false, fmt.Errorf("decode tag %s: %w", h, err)
+			}
+			h = tag.Target
+		default:
+			return plumbing.ZeroHash, false, nil
+		}
+	}
+	return plumbing.ZeroHash, false, errors.New("tag chain is too deep")
+}
+
+// humanBytes formats n in binary units with one decimal, such as "5.6 MiB".
+func humanBytes(n int64) string {
+	const unit = 1024
+	if n < unit {
+		return fmt.Sprintf("%d B", n)
+	}
+	div, exp := int64(unit), 0
+	for m := n / unit; m >= unit; m /= unit {
+		div *= unit
+		exp++
+	}
+	return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp])
+}