perf(storage/tigris): cap zstd encoder concurrency

cmd/objgitd/main.go

@@ -49,6 +49,8 @@ var (
 	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")
 
+	encoderConcurrency = flag.Int("encoder-concurrency", 4, "how many concurrent zstd EncodeAll calls the process-wide pack encoder serves; each state retains a match-history buffer, so this caps encoder heap that would otherwise scale with GOMAXPROCS. Set with the concurrent push cap in mind")
+
 	maxConcurrentPushes = flag.Int("max-concurrent-pushes", 4, "pushes allowed to unpack a packfile at the same time; each one costs roughly 400 MiB of resident set for a large repository, so this is what bounds memory under concurrent pushes; 0 disables the limit")
 	pushQueueTimeout    = flag.Duration("push-queue-timeout", 2*time.Minute, "how long a push waits for a slot before it fails")
 )
@@ -87,6 +89,11 @@ func main() {
 	// Route s3fs S3 round-trips into Prometheus before any filesystem use.
 	s3fs.SetMetricsObserver(metrics.ObserveS3)
 
+	// Cap the process-wide pack encoder's concurrency before the first encode.
+	// Left at its GOMAXPROCS default, each internal state retains a
+	// match-history buffer and the retained floor scales with core count.
+	tigris.SetEncoderConcurrency(*encoderConcurrency)
+
 	rawClient, err := tstorage.New(ctx)
 	if err != nil {
 		slog.Error("can't create Tigris storage client", "err", err)

internal/storage/tigris/compress.go

@@ -95,11 +95,39 @@ var (
 	payloadOnce sync.Once
 )
 
+// encoderConcurrency caps how many EncodeAll calls the encoder() singleton
+// serves at once. zstd.NewWriter defaults WithEncoderConcurrency to GOMAXPROCS,
+// and every internal encoder state allocates its own match-history buffer (tens
+// of megabytes, never released), so the default puts a permanent floor under
+// retained heap that scales with core count and not with real demand. The value
+// that matters is "expected simultaneous pushes", not "core count"; 4 matches
+// deltaScanWorkers in spirit — a bound picked for resource reasons. Revisit it
+// alongside the push cap that actually bounds demand.
+//
+// A package-level variable with a setter, not a Storer option: the encoder is a
+// process-wide singleton, so SetEncoderConcurrency must be called from main
+// before the first encode. Reads are not synchronized because the setter runs
+// once at startup, before any EncodeAll.
+var encoderConcurrency = 4
+
+// SetEncoderConcurrency sets how many EncodeAll calls the encoder() singleton
+// serves at once. Call it once from main before the first encode; values below
+// 1 are clamped to 1.
+func SetEncoderConcurrency(n int) {
+	if n < 1 {
+		n = 1
+	}
+	encoderConcurrency = n
+}
+
 func encoder() *zstd.Encoder {
 	zstdEncOnce.Do(func() {
 		// Errors here are impossible with a nil writer and valid options; the
 		// option list is a compile-time constant.
-		zstdEnc, _ = zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault))
+		zstdEnc, _ = zstd.NewWriter(nil,
+			zstd.WithEncoderLevel(zstd.SpeedDefault),
+			zstd.WithEncoderConcurrency(encoderConcurrency),
+		)
 	})
 	return zstdEnc
 }
@@ -114,7 +142,15 @@ func encoder() *zstd.Encoder {
 var streamEncPool = sync.Pool{
 	New: func() any {
 		// Same error-impossibility reasoning as encoder() above.
-		zw, _ := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault))
+		//
+		// WithEncoderConcurrency(1) because a streaming copy owns its encoder for
+		// the duration of one object, as the comment above states: the extra
+		// states a default GOMAXPROCS encoder would allocate can never be used
+		// concurrently by a single owner, so they are pure retained-heap floor.
+		zw, _ := zstd.NewWriter(nil,
+			zstd.WithEncoderLevel(zstd.SpeedDefault),
+			zstd.WithEncoderConcurrency(1),
+		)
 		return zw
 	},
 }