fix(controller): bound proxy upload file-cache charges

controller/docker_socket_proxy.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"golang.org/x/sys/unix"
 )
 
 const dockerProxySocket = "/run/ci-fleet/locks/docker.sock"
@@ -30,6 +31,7 @@ const dockerProxyHealth = "ci-fleet-docker-socket-proxy-v1\n"
 const dockerBuildContextLimit = 1 << 30
 const dockerBuildUploadLimit = 2
 const dockerMetadataBodyLimit = 1 << 20
+const dockerSpoolCacheWindow = 4 << 20
 
 var dockerAPIVersion = regexp.MustCompile(`^/v[0-9]+\.[0-9]+/`)
 
@@ -308,15 +310,79 @@ func (body *dockerProxyBody) Close() error {
 }
 
 type dockerBuildSpool struct {
-	*os.File
-	closed sync.Once
-	slots  chan struct{}
-	err    error
+	file    *os.File
+	offset  int64
+	dropped int64
+	closed  sync.Once
+	slots   chan struct{}
+	err     error
+}
+
+func (spool *dockerBuildSpool) requireDisk() error {
+	var filesystem unix.Statfs_t
+	if err := unix.Fstatfs(int(spool.file.Fd()), &filesystem); err != nil {
+		return err
+	}
+	if filesystem.Type == unix.TMPFS_MAGIC || filesystem.Type == unix.RAMFS_MAGIC {
+		return errors.New("Docker upload spool requires disk-backed temporary storage")
+	}
+	return nil
+}
+
+func (spool *dockerBuildSpool) Write(buffer []byte) (int, error) {
+	count, err := spool.file.Write(buffer)
+	spool.offset += int64(count)
+	if err == nil && spool.offset-spool.dropped >= dockerSpoolCacheWindow {
+		err = spool.releaseWritten()
+	}
+	return count, err
+}
+
+func (spool *dockerBuildSpool) releaseWritten() error {
+	// Dirty and reread file pages count against the proxy's memory cgroup too.
+	if err := unix.Fdatasync(int(spool.file.Fd())); err != nil {
+		return err
+	}
+	return spool.releaseCache()
+}
+
+func (spool *dockerBuildSpool) releaseCache() error {
+	length := spool.offset - spool.dropped
+	if length == 0 {
+		return nil
+	}
+	if err := unix.Fadvise(int(spool.file.Fd()), spool.dropped, length, unix.FADV_DONTNEED); err != nil {
+		return err
+	}
+	spool.dropped = spool.offset
+	return nil
+}
+
+func (spool *dockerBuildSpool) rewind() error {
+	if err := spool.releaseWritten(); err != nil {
+		return err
+	}
+	if _, err := spool.file.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
+	spool.offset, spool.dropped = 0, 0
+	return nil
+}
+
+func (spool *dockerBuildSpool) Read(buffer []byte) (int, error) {
+	count, err := spool.file.Read(buffer)
+	spool.offset += int64(count)
+	if spool.offset-spool.dropped >= dockerSpoolCacheWindow || errors.Is(err, io.EOF) {
+		if cacheErr := spool.releaseCache(); cacheErr != nil {
+			err = errors.Join(err, cacheErr)
+		}
+	}
+	return count, err
 }
 
 func (spool *dockerBuildSpool) Close() error {
 	spool.closed.Do(func() {
-		spool.err = spool.File.Close()
+		spool.err = spool.file.Close()
 		<-spool.slots
 	})
 	return spool.err
@@ -375,14 +441,18 @@ func newDockerSocketProxy(upstream, lockPath, bootID string) (http.Handler, *htt
 					http.Error(w, err.Error(), http.StatusServiceUnavailable)
 					return
 				}
-				spool := &dockerBuildSpool{File: file, slots: buildSlots}
+				spool := &dockerBuildSpool{file: file, slots: buildSlots}
 				defer spool.Close()
 				// Unlink before writing so a killed proxy cannot leave spooled contexts behind.
 				if err := os.Remove(file.Name()); err != nil {
 					http.Error(w, err.Error(), http.StatusServiceUnavailable)
 					return
 				}
-				length, err := io.Copy(file, http.MaxBytesReader(w, request.Body, dockerBuildContextLimit))
+				if err := spool.requireDisk(); err != nil {
+					http.Error(w, err.Error(), http.StatusServiceUnavailable)
+					return
+				}
+				length, err := io.Copy(spool, http.MaxBytesReader(w, request.Body, dockerBuildContextLimit))
 				if err != nil {
 					var tooLarge *http.MaxBytesError
 					if errors.As(err, &tooLarge) {
@@ -392,7 +462,7 @@ func newDockerSocketProxy(upstream, lockPath, bootID string) (http.Handler, *htt
 					http.Error(w, err.Error(), http.StatusBadRequest)
 					return
 				}
-				if _, err := file.Seek(0, io.SeekStart); err != nil {
+				if err := spool.rewind(); err != nil {
 					http.Error(w, err.Error(), http.StatusServiceUnavailable)
 					return
 				}

controller/docker_socket_proxy_spool_test.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestDockerBuildSpoolRetainsBytesAcrossWriteAndReadCacheWindows(t *testing.T) {
+	file, err := os.CreateTemp(t.TempDir(), "spool-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	slots := make(chan struct{}, 1)
+	slots <- struct{}{}
+	spool := &dockerBuildSpool{file: file, slots: slots}
+	defer spool.Close()
+	if err := spool.requireDisk(); err != nil {
+		t.Fatal(err)
+	}
+	input := bytes.Repeat([]byte("exact-upload-content\n"), dockerSpoolCacheWindow/10+17)
+	want := sha256.Sum256(input)
+	if count, err := io.Copy(spool, bytes.NewReader(input)); err != nil || count != int64(len(input)) {
+		t.Fatalf("spooling lost bytes: count=%d, error=%v", count, err)
+	}
+	if err := spool.rewind(); err != nil {
+		t.Fatal(err)
+	}
+	actual := sha256.New()
+	if count, err := io.Copy(actual, spool); err != nil || count != int64(len(input)) {
+		t.Fatalf("forwarding lost bytes: count=%d, error=%v", count, err)
+	}
+	if !bytes.Equal(actual.Sum(nil), want[:]) {
+		t.Fatal("cache release changed the forwarded payload")
+	}
+	if err := spool.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if err := spool.Close(); err != nil || len(slots) != 0 {
+		t.Fatalf("closing must release exactly one upload slot: %v", err)
+	}
+}
+
+func TestDockerBuildSpoolRefusesFailedCachePreparation(t *testing.T) {
+	file, err := os.CreateTemp(t.TempDir(), "closed-spool-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := file.Close(); err != nil {
+		t.Fatal(err)
+	}
+	spool := &dockerBuildSpool{file: file}
+	if err := spool.requireDisk(); err == nil {
+		t.Fatal("unreadable filesystem identity was accepted")
+	}
+	if err := spool.rewind(); err == nil {
+		t.Fatal("failed file synchronization must not reach forwarding")
+	}
+	if _, err := spool.Write([]byte("upload")); err == nil {
+		t.Fatal("failed spool write was accepted")
+	}
+	if _, err := spool.Read(make([]byte, 8)); err == nil {
+		t.Fatal("failed spool read was accepted")
+	}
+}
+
+func TestDockerBuildSpoolRefusesMemoryBackedTemporaryFiles(t *testing.T) {
+	file, err := os.CreateTemp("/dev/shm", "proxy-memory-spool-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(file.Name())
+	defer file.Close()
+	spool := &dockerBuildSpool{file: file}
+	if err := spool.requireDisk(); err == nil {
+		t.Fatal("memory-backed spooling cannot fit a full upload within the proxy memory limit")
+	}
+}
+
+func TestDockerSocketProxyRefusesMemoryBackedSpoolBeforeDispatch(t *testing.T) {
+	directory, err := os.MkdirTemp("/dev/shm", "proxy-spool-refusal-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(directory)
+	dispatched := make(chan struct{}, 1)
+	fixture := newSocketProxyTest(t, http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		dispatched <- struct{}{}
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Setenv("TMPDIR", directory)
+	response, err := fixture.client.Post("http://docker/images/load", "application/x-tar", strings.NewReader("fixture"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	body, err := io.ReadAll(response.Body)
+	if err != nil || response.StatusCode != http.StatusServiceUnavailable || !strings.Contains(string(body), "disk-backed") {
+		t.Fatalf("memory-backed spool was not refused: %d %s, %v", response.StatusCode, body, err)
+	}
+	select {
+	case <-dispatched:
+		t.Fatal("memory-backed spool reached Docker")
+	default:
+	}
+}

controller/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/docker/docker v28.5.2+incompatible
 	github.com/google/uuid v1.6.0
 	github.com/moby/docker-image-spec v1.3.1
+	golang.org/x/sys v0.49.0
 )
 
 require (
@@ -35,5 +36,4 @@ require (
 	go.opentelemetry.io/otel/metric v1.47.0 // indirect
 	go.opentelemetry.io/otel/trace v1.47.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.11.0 // indirect
-	golang.org/x/sys v0.49.0 // indirect
 )

docs/HOST-MAINTENANCE.md

@@ -129,6 +129,17 @@ image removal, and builder pruning participate in proxy protocol `v2` alongside
 builds and container reference mutations. Storage removal defers with older
 proxies. Image-load uploads use the same 1 GiB spool limit as build contexts.
 Image imports use that bounded spool too and share its two upload slots.
+Upload spools synchronize written file data and discard its cached pages in
+4 MiB windows. Forwarding also discards consumed pages. File-cache charges
+belong to the proxy's memory cgroup, so an unlinked disk file alone does not
+keep a large upload within its unchanged 128 MiB memory limit. Synchronization
+or cache-advice failures during spooling refuse the request before forwarding.
+A cache-advice error during forwarding retains the existing uncertainty guard.
+The body limit and two upload slots are unchanged.
+The temporary directory must have disk backing. Memory-backed `tmpfs` or `ramfs`
+spools are refused before forwarding because their data cannot be evicted from
+the memory limit as a disk page can. Do not move spools to a memory filesystem
+to avoid synchronization costs.
 Registry pulls and imports from remote URLs bypass upload slots while retaining
 the maintenance lock. Other guarded request bodies have a 1 MiB metadata limit;
 oversized bodies receive HTTP 413 before reaching Docker.

scripts/install-worker-controller.sh

@@ -824,7 +824,7 @@ PY
 }
 
 runtime_release_complete() {
-  local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required stored_digest actual_digest policy_script
+  local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required stored_digest actual_digest policy_script enabled_capabilities
   local -a capability_args=()
   [[ -d "$path" && -f "$path/.ci-fleet-engine-ref" && -f "$path/.ci-fleet-tree-sha256" && -f "$path/deploy/compose.yaml" ]] || return 1
   [[ -x "$path/scripts/preflight.sh" && -x "$path/scripts/healthcheck.sh" && -x "$path/scripts/cleanup.sh" ]] || return 1
@@ -834,14 +834,10 @@ runtime_release_complete() {
     [[ "$require_status" != 1 ]] || capability_args+=(--require-status-reporting)
     python3 "$repo_root/scripts/desired_state.py" validate-engine-capabilities \
       --manifest "$path/engine-capabilities.json" "${capability_args[@]}" >/dev/null || return 1
+    enabled_capabilities=$(jq -c '.capabilities | to_entries[] | select(.value == true) | .key' "$path/engine-capabilities.json") || return 1
     for required in docker_network_policy_config:apply-docker-network-policy.sh docker_network_policy_adapter:docker-network-policy-adapter.sh reproducible_managed_images:build-managed-images.sh portable_managed_image_ids:managed_image_identity.py apt_cache:apt_cache.py; do
       policy_script=${required#*:}
-      if python3 - "$path/engine-capabilities.json" "${required%%:*}" <<'PY'
-import json, sys
-value = json.load(open(sys.argv[1], encoding="utf-8"))
-raise SystemExit(value.get("capabilities", {}).get(sys.argv[2]) is not True)
-PY
-      then
+      if grep -Fxq -- "\"${required%%:*}\"" <<<"$enabled_capabilities"; then
         [[ -f "$path/scripts/$policy_script" && ! -L "$path/scripts/$policy_script" ]] || return 1
         [[ "$policy_script" == *.py || -x "$path/scripts/$policy_script" ]] || return 1
       fi

scripts/test-install-worker-controller.sh

@@ -1930,7 +1930,7 @@ retry_result=$tmp/interrupted-retry-result.json
 : >"$retry_result"
 FAKE_PAUSE_SYSTEMCTL_ONCE=$retry_pause CI_FLEET_TRANSACTION_RESULT_FD=7 "$installer" --upgrade "${base_args[@]}" --ref "$interrupted_ref" 7>"$retry_result" >"$tmp/interrupted-retry.out" 2>&1 &
 retry_pid=$!
-wait_for_file "$retry_pause.entered" 'interrupted policy retry did not reach nested activation'
+(wait_for_file "$retry_pause.entered" 'interrupted policy retry did not reach nested activation') || fail "interrupted policy retry did not reach nested activation: $(<"$tmp/interrupted-retry.out")"
 for repaired_checkpoint_target in "$stale_checkpoint_release" "$stale_checkpoint_manager"; do
   if find "$repaired_checkpoint_target" \( -type d -o -type f \) -perm /022 -print -quit | grep -q .; then
     fail "pending checkpoint target remained permission-unsafe: $repaired_checkpoint_target"

scripts/test_local_validation.py

@@ -452,12 +452,44 @@ def test_successful_leader_allows_descendant_cleanup(self):
                 run(["python3", "-c", leader], root, os.environ.copy(), root / "test.log", 5)
             self.assertEqual((root / "cleaned").read_text(), "done")
 
+    def test_proxy_memory_timeout_unwinds_owned_fixture_through_shell(self):
+        for logs_status in (0, 1):
+            with self.subTest(logs_status=logs_status), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                docker = root / "docker"
+                docker.write_text('''#!/usr/bin/python3
+import json, os, sys, time
+from pathlib import Path
+args = sys.argv[1:]
+with Path(os.environ["PROXY_COMMANDS"]).open("a") as log:
+    log.write(json.dumps(args) + "\\n")
+if args[:2] == ["run", "-d"]:
+    print("fixture-proxy")
+if args[:2] == ["inspect", "--format"]:
+    time.sleep(60)
+if args[:1] == ["logs"]:
+    sys.exit(int(os.environ["PROXY_LOGS_STATUS"]))
+''')
+                docker.chmod(0o755)
+                commands = root / "commands.jsonl"
+                env = os.environ | {"PATH": str(root) + ":" + os.environ["PATH"],
+                                    "PROXY_COMMANDS": str(commands), "PROXY_LOGS_STATUS": str(logs_status)}
+                command = ["bash", "-c", 'python3 "$1" --image fixture-image; exit "$?"',
+                           "bash", str(ROOT / "scripts/test_proxy_memory.py")]
+                with self.assertRaises(subprocess.TimeoutExpired):
+                    run(command, ROOT, env, root / "fixture.log", 2)
+                recorded = [json.loads(line) for line in commands.read_text().splitlines()]
+                self.assertTrue(any(args[:2] == ["run", "-d"] for args in recorded))
+                self.assertIn(["logs", "fixture-proxy"], recorded)
+                self.assertIn(["rm", "-f", "fixture-proxy"], recorded)
+
     def test_container_stage_propagates_cleanup_failure(self):
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
             scripts = root / "scripts"; scripts.mkdir()
             (scripts / "test_apt_cache_e2e.py").write_text(
                 "import os, sys\nsys.exit(int(os.environ['TEST_STAGE_STATUS']))\n")
+            (scripts / "test_proxy_memory.py").write_text("")
             (scripts / "scan_committed_secrets.py").write_text("")
             for name in ("test-cleanup-e2e.sh", "cleanup.sh"):
                 path = scripts / name