feat(runtime): opt-in per-job memory budget; stop leaking dind parent cgroups

cmd/ephemerd/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/ephpm/ephemerd/pkg/dind"
 	"github.com/ephpm/ephemerd/pkg/github"
 	"github.com/ephpm/ephemerd/pkg/imagegc"
+	"github.com/ephpm/ephemerd/pkg/jobcgroup"
 	"github.com/ephpm/ephemerd/pkg/metrics"
 	"github.com/ephpm/ephemerd/pkg/networking"
 	"github.com/ephpm/ephemerd/pkg/providers"
@@ -361,6 +362,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP
 			DindAllowPrivileged: cfg.Dind.ResolvedAllowPrivileged(),
 			Rlimits:             cfg.Runtime.Rlimits.Resolved(),
 			PidsLimit:           cfg.Runtime.ResolvedPidsLimit(),
+			MemoryLimit:         memoryLimit(cfg.Runtime),
 			AllowNewPrivileges:  cfg.Runtime.ResolvedAllowNewPrivileges(),
 			LinuxRuntime:        cfg.Runner.Linux.ContainerdRuntime(),
 			Network:             net,
@@ -685,6 +687,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP
 		CacheProxyMounts:    cacheProxyMounts,
 		Rlimits:             cfg.Runtime.Rlimits.Resolved(),
 		PidsLimit:           cfg.Runtime.ResolvedPidsLimit(),
+		MemoryLimit:         memoryLimit(cfg.Runtime),
 		AllowNewPrivileges:  cfg.Runtime.ResolvedAllowNewPrivileges(),
 		LinuxRuntime:        cfg.Runner.Linux.ContainerdRuntime(),
 		Network:             net,
@@ -1131,6 +1134,35 @@ func buildkitGCConfig(cfg *config.Config) buildkit.GCConfig {
 	}
 }
 
+// memoryLimit resolves [runtime] memory_limit_mb / memory_high_percent into
+// the job cgroup budget the runtime and dind apply.
+func memoryLimit(r config.RuntimeConfig) jobcgroup.Limits {
+	maxBytes, highBytes := r.ResolvedMemoryLimit()
+	return jobcgroup.Limits{MaxBytes: maxBytes, HighBytes: highBytes}
+}
+
+// jobCgroupSweepGrace is how long an empty job parent cgroup must sit
+// unchanged before the periodic sweep removes it. Teardown removes a job's
+// own parent; the sweep is the backstop for jobs that never reached teardown
+// (daemon crash, host reboot) and for the backlog older daemons left.
+const jobCgroupSweepGrace = 10 * time.Minute
+
+// sweepJobCgroups runs the leaked job cgroup sweep, skipping every job rt is
+// still running or provisioning. If the live set cannot be read the pass is
+// skipped: removing a live job's parent is worse than leaking one a tick
+// longer.
+func sweepJobCgroups(ctx context.Context, rt *runtime.Runtime, log *slog.Logger) {
+	var live map[string]bool
+	if rt != nil {
+		var err error
+		if live, err = rt.LiveJobIDs(ctx); err != nil {
+			log.Debug("job cgroup sweep skipped: cannot list live jobs", "error", err)
+			return
+		}
+	}
+	jobcgroup.Sweep(jobCgroupSweepGrace, live, log)
+}
+
 // runNodeDiskSweeper is the periodic half of node disk hygiene, on one
 // timer: an orphan sweep, a sweep of dead jobs' BuildKit export records,
 // and a disk-pressure image collection pass.
@@ -1146,6 +1178,9 @@ func buildkitGCConfig(cfg *config.Config) buildkit.GCConfig {
 func runNodeDiskSweeper(ctx context.Context, gc *imagegc.Collector, rt *runtime.Runtime, c *containerdclient.Client, interval time.Duration, repair brokenChainRepair, log *slog.Logger) {
 	log = log.With("component", "node-disk-sweeper", "interval", interval)
 	log.Info("starting node disk sweeper")
+	startCtx, startCancel := context.WithTimeout(ctx, time.Minute)
+	sweepJobCgroups(startCtx, rt, log)
+	startCancel()
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 	for {
@@ -1169,6 +1204,7 @@ func runNodeDiskSweeper(ctx context.Context, gc *imagegc.Collector, rt *runtime.
 				}
 			}
 			sweepDeadBuildRecords(passCtx, c, log)
+			sweepJobCgroups(passCtx, rt, log)
 			// Correctness repair, not capacity policy — so it runs on
 			// every tick whether or not image GC is enabled, and before
 			// the collector, which would otherwise plan against records

pkg/config/config.go

@@ -362,6 +362,29 @@ type RuntimeConfig struct {
 	// others and a busy neighbour cannot fail a build.
 	PidsLimit int64 `toml:"pids_limit"`
 
+	// MemoryLimitMB is the memory budget for one JOB: the runner container
+	// and every sibling it starts through dind, together, share one cgroup
+	// with this memory.max. 0 (unset) means no limit, which is the
+	// historical behaviour.
+	//
+	// It is a per-job budget, not per-container, because the heavy work
+	// in a `container:` job runs inside a dind sibling: a per-container
+	// limit would let one job spend N × the limit. When the budget is hit
+	// the kernel OOM-kills inside that job only — usually the largest
+	// process (a linker), failing the step — instead of the host-wide OOM
+	// that on sharks (2026-10-23 03:00Z, 40 GB, 5 slots) killed processes
+	// in three concurrent jobs at once.
+	MemoryLimitMB int64 `toml:"memory_limit_mb"`
+
+	// MemoryHighPercent sets memory.high as a percentage of MemoryLimitMB.
+	// Above memory.high the kernel reclaims the job's page cache and
+	// throttles it before memory.max is reached, so a build that only
+	// looks big because of cache (target/ dirs, registry layers) slows
+	// down rather than getting killed. 0 (unset) means
+	// DefaultMemoryHighPercent; valid range 50-100, where 100 disables the
+	// soft limit. Ignored when MemoryLimitMB is 0.
+	MemoryHighPercent int `toml:"memory_high_percent"`
+
 	// AllowNewPrivileges controls whether a process in the runner
 	// container may gain privileges through execve — i.e. whether setuid
 	// binaries and file capabilities take effect. When true the OCI spec
@@ -502,6 +525,46 @@ func (r RuntimeConfig) ResolvedPidsLimit() int64 {
 	}
 }
 
+// DefaultMemoryHighPercent is memory.high as a share of [runtime]
+// memory_limit_mb when memory_high_percent is unset.
+const DefaultMemoryHighPercent = 90
+
+// minMemoryLimitMB is the smallest per-job budget accepted. Anything below
+// it cannot hold the runner itself, and is far more likely a units mistake
+// (GB written as MB) than a deliberate setting.
+const minMemoryLimitMB = 512
+
+// ResolvedMemoryLimit returns the job cgroup's memory.max and memory.high in
+// bytes. Both are 0 when no limit is configured; high is 0 (no soft limit)
+// when memory_high_percent is 100.
+func (r RuntimeConfig) ResolvedMemoryLimit() (maxBytes, highBytes int64) {
+	if r.MemoryLimitMB <= 0 {
+		return 0, 0
+	}
+	maxBytes = r.MemoryLimitMB << 20
+	pct := r.MemoryHighPercent
+	if pct == 0 {
+		pct = DefaultMemoryHighPercent
+	}
+	if pct >= 100 {
+		return maxBytes, 0
+	}
+	return maxBytes, maxBytes * int64(pct) / 100
+}
+
+func (r RuntimeConfig) validate() error {
+	if r.MemoryLimitMB < 0 {
+		return fmt.Errorf("[runtime] memory_limit_mb = %d: must be 0 (no limit) or at least %d", r.MemoryLimitMB, minMemoryLimitMB)
+	}
+	if r.MemoryLimitMB > 0 && r.MemoryLimitMB < minMemoryLimitMB {
+		return fmt.Errorf("[runtime] memory_limit_mb = %d: below the %d MB floor (the value is in megabytes — 16 GiB is 16384)", r.MemoryLimitMB, minMemoryLimitMB)
+	}
+	if p := r.MemoryHighPercent; p != 0 && (p < 50 || p > 100) {
+		return fmt.Errorf("[runtime] memory_high_percent = %d: must be between 50 and 100", p)
+	}
+	return nil
+}
+
 // defaultNproc is RLIMIT_NPROC when [runtime.rlimits] nproc is unset. It is
 // deliberately effectively unlimited: NPROC is enforced per UID across the
 // host, not per container, so as a per-job limit it only ever made
@@ -2038,6 +2101,10 @@ func (c *Config) validate() error {
 		return err
 	}
 
+	if err := c.Runtime.validate(); err != nil {
+		return err
+	}
+
 	if err := c.Runner.Linux.validate(); err != nil {
 		return err
 	}

pkg/config/config_test.go

@@ -1,4 +1,4 @@
-package config
+package config
 
 import (
 	"bytes"
@@ -2518,3 +2518,40 @@ func TestRuntimeConfig_ResolvedPidsLimit(t *testing.T) {
 		}
 	}
 }
+
+func TestRuntimeConfig_ResolvedMemoryLimit(t *testing.T) {
+	for _, tc := range []struct {
+		name              string
+		in                RuntimeConfig
+		wantMax, wantHigh int64
+	}{
+		{"unset: no limit", RuntimeConfig{}, 0, 0},
+		{"default soft limit", RuntimeConfig{MemoryLimitMB: 16384}, 16 << 30, (16 << 30) * DefaultMemoryHighPercent / 100},
+		{"explicit percent", RuntimeConfig{MemoryLimitMB: 1000, MemoryHighPercent: 50}, 1000 << 20, 500 << 20},
+		{"100 percent: no soft limit", RuntimeConfig{MemoryLimitMB: 2048, MemoryHighPercent: 100}, 2048 << 20, 0},
+	} {
+		gotMax, gotHigh := tc.in.ResolvedMemoryLimit()
+		if gotMax != tc.wantMax || gotHigh != tc.wantHigh {
+			t.Errorf("%s: ResolvedMemoryLimit() = (%d, %d), want (%d, %d)", tc.name, gotMax, gotHigh, tc.wantMax, tc.wantHigh)
+		}
+	}
+}
+
+func TestRuntimeConfig_ValidateMemoryLimit(t *testing.T) {
+	for _, tc := range []struct {
+		in      RuntimeConfig
+		wantErr bool
+	}{
+		{RuntimeConfig{}, false},
+		{RuntimeConfig{MemoryLimitMB: 16384}, false},
+		{RuntimeConfig{MemoryLimitMB: 16384, MemoryHighPercent: 85}, false},
+		{RuntimeConfig{MemoryLimitMB: 16}, true}, // GB typed as MB
+		{RuntimeConfig{MemoryLimitMB: -1}, true},
+		{RuntimeConfig{MemoryLimitMB: 4096, MemoryHighPercent: 40}, true},
+		{RuntimeConfig{MemoryLimitMB: 4096, MemoryHighPercent: 101}, true},
+	} {
+		if err := tc.in.validate(); (err != nil) != tc.wantErr {
+			t.Errorf("validate(%+v) = %v, wantErr %v", tc.in, err, tc.wantErr)
+		}
+	}
+}

pkg/dind/containers.go

@@ -526,6 +526,10 @@ func (s *Server) handleContainerCreate(w http.ResponseWriter, r *http.Request) {
 	// host UID with every other job's containers, so RLIMIT_NPROC cannot
 	// bound it per job.
 	opts = append(opts, pidsLimitOpts(s.pidsLimit)...)
+	// The job's memory budget lives on the job parent cgroup this sibling is
+	// created under; (re)write it now so it holds even if the periodic sweep
+	// removed an empty parent between two of the job's containers.
+	opts = append(opts, jobMemoryOpts(s.jobID, s.memoryLimit, s.log)...)
 
 	if req.HostConfig != nil {
 		// Privileged mode: all capabilities, all devices, disable seccomp/apparmor,

pkg/dind/dind.go

@@ -32,6 +32,7 @@ import (
 	"github.com/containerd/containerd/v2/core/images"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
 	"github.com/ephpm/ephemerd/pkg/buildkit"
+	"github.com/ephpm/ephemerd/pkg/jobcgroup"
 	"github.com/ephpm/ephemerd/pkg/networking"
 	"github.com/ephpm/ephemerd/pkg/registrymirror"
 )
@@ -72,6 +73,7 @@ type Server struct {
 	runnerNetNS     string           // path to runner container's net namespace; used to install DNAT rules for port bindings
 	allowPrivileged bool             // gate for docker run --privileged / --cap-add; see config.DindConfig.AllowPrivileged
 	pidsLimit       int64            // cgroup pids.max for siblings; 0 = no limit
+	memoryLimit     jobcgroup.Limits // the job's memory budget, written on the job parent cgroup
 
 	// stager publishes every job-supplied bind source at a path under
 	// <DataDir>/dind-binds/<JobID>/ that only root can reach, so the path
@@ -191,6 +193,13 @@ type Config struct {
 	// config.RuntimeConfig.PidsLimit.
 	PidsLimit int64
 
+	// MemoryLimit is the job's memory budget, shared by the runner and every
+	// sibling: siblings already live under the job parent cgroup (it is the
+	// cgroup containerd derives from the job's namespace), so applying the
+	// budget means writing it on that parent before each sibling starts.
+	// Zero means no limit. See pkg/jobcgroup.
+	MemoryLimit jobcgroup.Limits
+
 	// RegistryMirror routes this job's image pulls through a LAN
 	// pull-through cache. Nil disables mirroring, leaving the pulls
 	// identical to what they were before the feature existed.
@@ -251,7 +260,8 @@ func New(cfg Config) (*Server, error) {
 		runnerNetNS:     cfg.RunnerNetNS,
 		allowPrivileged: cfg.AllowPrivileged,
 		pidsLimit:       cfg.PidsLimit,
-		stager:         newBindStager(cfg.DataDir, cfg.JobID, cfg.Log),
+		memoryLimit:     cfg.MemoryLimit,
+		stager:          newBindStager(cfg.DataDir, cfg.JobID, cfg.Log),
 		mirror:          cfg.RegistryMirror,
 		resultsURL:      cfg.ResultsURL,
 		log:             cfg.Log.With("component", "dind", "job_id", cfg.JobID),

pkg/dind/jobcgroup_prefix_test.go

@@ -0,0 +1,22 @@
+package dind
+
+import (
+	"testing"
+
+	"github.com/ephpm/ephemerd/pkg/jobcgroup"
+)
+
+// TestJobCgroupPrefixIsTheDindNamespacePrefix pins the coupling the job
+// memory budget and the leaked-cgroup sweep both rely on: containerd puts a
+// sibling in cgroup "/<namespace>/<id>", so the job parent cgroup is named
+// after the job's dind namespace. If either prefix changes alone, siblings
+// escape the job budget and the sweep stops matching what jobs leave behind.
+func TestJobCgroupPrefixIsTheDindNamespacePrefix(t *testing.T) {
+	if jobcgroup.Prefix != DindNamespacePrefix {
+		t.Fatalf("jobcgroup.Prefix = %q, DindNamespacePrefix = %q: they must match", jobcgroup.Prefix, DindNamespacePrefix)
+	}
+	s := &Server{jobNamespace: DindNamespacePrefix + "job-1"}
+	if "/"+s.jobNamespace != jobcgroup.Dir("job-1") {
+		t.Fatalf("job namespace %q does not name the job cgroup %q", s.jobNamespace, jobcgroup.Dir("job-1"))
+	}
+}

pkg/dind/pidslimit_linux.go

@@ -4,9 +4,11 @@ package dind
 
 import (
 	"context"
+	"log/slog"
 
 	"github.com/containerd/containerd/v2/core/containers"
 	"github.com/containerd/containerd/v2/pkg/oci"
+	"github.com/ephpm/ephemerd/pkg/jobcgroup"
 	ocispec "github.com/opencontainers/runtime-spec/specs-go"
 )
 
@@ -31,3 +33,37 @@ func pidsLimitOpts(limit int64) []oci.SpecOpts {
 		},
 	}
 }
+
+// jobMemoryOpts applies the job's memory budget to a sibling by placing it
+// under the job parent cgroup and (re)writing the parent's limits first. The
+// default spec already puts it there — containerd derives CgroupsPath from
+// the job's dind namespace, which is the parent's name — but the path is set
+// explicitly so the budget cannot silently stop applying if that derivation
+// ever changes. No budget adds nothing.
+//
+// Fail-open, like the runner side: a parent that cannot be limited leaves the
+// sibling on its default cgroup, unlimited, rather than failing the create.
+func jobMemoryOpts(jobID string, limits jobcgroup.Limits, log *slog.Logger) []oci.SpecOpts {
+	if !limits.Enabled() || jobID == "" {
+		return nil
+	}
+	return []oci.SpecOpts{
+		func(_ context.Context, _ oci.Client, c *containers.Container, s *oci.Spec) error {
+			if err := ensureJobCgroup(jobcgroup.Root, jobID, limits); err != nil {
+				log.Warn("job memory limit not applied to sibling; it runs unlimited", "error", err)
+				return nil
+			}
+			if s.Linux == nil {
+				s.Linux = &ocispec.Linux{}
+			}
+			if c != nil && c.ID != "" {
+				s.Linux.CgroupsPath = jobcgroup.Dir(jobID) + "/" + c.ID
+			}
+			return nil
+		},
+	}
+}
+
+// ensureJobCgroup is jobcgroup.Ensure, a var so tests can run the spec
+// option without a real cgroupfs.
+var ensureJobCgroup = jobcgroup.Ensure

pkg/dind/pidslimit_linux_test.go

@@ -0,0 +1,56 @@
+//go:build linux
+
+package dind
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"testing"
+
+	"github.com/containerd/containerd/v2/core/containers"
+	"github.com/containerd/containerd/v2/pkg/oci"
+	"github.com/ephpm/ephemerd/pkg/jobcgroup"
+	ocispec "github.com/opencontainers/runtime-spec/specs-go"
+)
+
+// TestJobMemoryOpts: a sibling of a job with a memory budget is placed under
+// the job parent cgroup, whose limits are (re)written first; with no budget,
+// or when the parent cannot be prepared, the sibling keeps its default.
+func TestJobMemoryOpts(t *testing.T) {
+	old := ensureJobCgroup
+	t.Cleanup(func() { ensureJobCgroup = old })
+	limits := jobcgroup.Limits{MaxBytes: 4 << 30, HighBytes: 3 << 30}
+
+	var ensured []string
+	ensureJobCgroup = func(_ string, jobID string, l jobcgroup.Limits) error {
+		if l != limits {
+			t.Errorf("Ensure got %+v, want %+v", l, limits)
+		}
+		ensured = append(ensured, jobID)
+		return nil
+	}
+	apply := func(opts []oci.SpecOpts) string {
+		spec := &oci.Spec{Linux: &ocispec.Linux{CgroupsPath: "/default/path"}}
+		for _, opt := range opts {
+			if err := opt(context.Background(), nil, &containers.Container{ID: "abc123"}, spec); err != nil {
+				t.Fatalf("opt: %v", err)
+			}
+		}
+		return spec.Linux.CgroupsPath
+	}
+
+	if got, want := apply(jobMemoryOpts("job-1", limits, slog.Default())), jobcgroup.Dir("job-1")+"/abc123"; got != want {
+		t.Errorf("CgroupsPath = %q, want %q", got, want)
+	}
+	if len(ensured) != 1 || ensured[0] != "job-1" {
+		t.Errorf("Ensure calls = %v, want one for job-1", ensured)
+	}
+	if opts := jobMemoryOpts("job-1", jobcgroup.Limits{}, slog.Default()); len(opts) != 0 {
+		t.Errorf("no budget produced %d opts, want none", len(opts))
+	}
+	ensureJobCgroup = func(string, string, jobcgroup.Limits) error { return errors.New("no memory controller") }
+	if got := apply(jobMemoryOpts("job-1", limits, slog.Default())); got != "/default/path" {
+		t.Errorf("CgroupsPath after a failed Ensure = %q, want the default kept (fail-open)", got)
+	}
+}