From c354095875eb0e0d07b0bf065dd9c5721e3b545e Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 11 Aug 2026 17:38:02 -0700 Subject: [PATCH 1/2] feat(proxies): cargo/crates pull-through cache, and fix the Go module proxy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Go module proxy has never once served a request. It binds to the CNI bridge gateway (10.88.0.1), but that address does not exist when the daemon starts: the bridge is created lazily by CNI on the first job container, and both cleanup() and cleanStaleBridge() delete it outright. So Listen fails with EADDRNOTAVAIL on every boot, the error is logged as a warning and swallowed, and GOPROXY is never injected into any container. Start() runs MkdirAll before the failing Listen, which is why cache/gomod exists on every node, empty, at exactly 4 KB. Three further defects would have kept it useless even with the bind fixed: EnvVars() advertised the bound address rather than the configured one; the ",direct" fallback only applies on 404/410, so a wedged proxy hard-failed builds instead of failing open; and the cleanup knob was dead code that wiped the cache on every shutdown regardless of configuration. - proxies: add Listen() with a wildcard fallback, so the gateway address becomes reachable as soon as CNI brings the bridge up. A real error (port in use) is still returned rather than masked. - go: advertise the configured address; use "|direct" so any proxy error falls through to upstream; honour cleanup=false. - cargo: new pull-through cache for the sparse index, .crate tarballs and rustup dist. The index is revalidated on a TTL with conditional GETs; tarballs are immutable and cached permanently. config.json is rewritten so dl points at the proxy while api still reaches the real registry. Fails open in three stages: not started, serve-stale, then a redirect to the origin. - runtime: support bind-mounting proxy-generated files into job containers. Cargo ignores CARGO_SOURCE_* and CARGO_REGISTRIES_* environment variables for source replacement — verified empirically, not assumed. Only a config file works, so the proxy generates .cargo/config.toml and mounts it read-only at the container root, relying on Cargo's ancestor-directory config search. That applies to any checkout path, leaves CARGO_HOME writable, and lets a repository's own config still win. No workflow changes are required. [cargo_proxy].cleanup defaults to false: wiping a pull-through cache on every restart is the bug diagnosed above. Also adds docs/arch/sccache-evaluation.md, which recommends a local-disk sccache scoped to cargo build (not docker build, to keep it disjoint from the BuildKit layer cache) with a hard size cap, sequenced after the disk-pressure GC lands. --- README.md | 10 + cmd/ephemerd/cache.go | 12 + cmd/ephemerd/main.go | 51 +- config.example.toml | 30 + docs/arch/sccache-evaluation.md | 197 +++++++ docs/cli/cache.md | 1 + docs/getting-started/configuration.md | 66 +++ pkg/config/config.go | 55 +- pkg/config/config_test.go | 86 ++- pkg/proxies/cargo/cache.go | 355 ++++++++++++ pkg/proxies/cargo/cache_test.go | 542 +++++++++++++++++++ pkg/proxies/cargo/cargoproxy.go | 752 ++++++++++++++++++++++++++ pkg/proxies/cargo/cargoproxy_test.go | 678 +++++++++++++++++++++++ pkg/proxies/go/goproxy.go | 36 +- pkg/proxies/listen.go | 79 +++ pkg/proxies/listen_test.go | 122 +++++ pkg/proxies/proxy.go | 26 + pkg/runtime/runtime.go | 53 +- 18 files changed, 3127 insertions(+), 24 deletions(-) create mode 100644 docs/arch/sccache-evaluation.md create mode 100644 pkg/proxies/cargo/cache.go create mode 100644 pkg/proxies/cargo/cache_test.go create mode 100644 pkg/proxies/cargo/cargoproxy.go create mode 100644 pkg/proxies/cargo/cargoproxy_test.go create mode 100644 pkg/proxies/listen.go create mode 100644 pkg/proxies/listen_test.go diff --git a/README.md b/README.md index 117bd969..11882065 100644 --- a/README.md +++ b/README.md @@ -389,6 +389,16 @@ enabled = true # run a GOPROXY on the bridge gateway # upstream = "https://proxy.golang.org" # upstream to fetch from on cache miss # cleanup = true # wipe cache on shutdown +# Cargo/crates caching proxy — pull-through cache for the crates.io sparse +# index, .crate tarballs, and rustup toolchains. No workflow changes needed. +[cargo_proxy] +enabled = true # run the Cargo proxy on the bridge gateway +# port = 8083 # default listen port +# upstream = "https://index.crates.io" # sparse registry index +# rustup_upstream = "https://static.rust-lang.org" +# index_ttl = "10m" # index revalidation (tarballs are immutable) +# cleanup = false # keep the cache across restarts + [log] level = "info" # debug, info, warn, error format = "text" # text or json diff --git a/cmd/ephemerd/cache.go b/cmd/ephemerd/cache.go index 2bc67a1e..16e84ed5 100644 --- a/cmd/ephemerd/cache.go +++ b/cmd/ephemerd/cache.go @@ -65,6 +65,18 @@ func managedCaches() []cacheEntry { Description: "Go module proxy cache (GOPROXY) served to job containers", LiveSafe: true, }, + { + Name: "cargo", + Rel: filepath.Join("cache", "cargo"), + Description: "Cargo proxy cache (crates.io sparse index, .crate tarballs, rustup toolchains) " + + "served to job containers", + // Live-safe: every entry is a pull-through copy of public + // registry content, so a running job that misses simply + // refetches it. The generated container config that jobs + // bind-mount deliberately lives OUTSIDE this dir (/cargo), + // so clearing the cache cannot yank it from a running job. + LiveSafe: true, + }, { Name: "buildkit", Rel: "buildkit", diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 17867408..6a84c0d3 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -27,6 +27,7 @@ import ( "github.com/ephpm/ephemerd/pkg/providers/gitea" githubProv "github.com/ephpm/ephemerd/pkg/providers/github" "github.com/ephpm/ephemerd/pkg/proxies" + cargoproxy "github.com/ephpm/ephemerd/pkg/proxies/cargo" goproxy "github.com/ephpm/ephemerd/pkg/proxies/go" "github.com/ephpm/ephemerd/pkg/runner" "github.com/ephpm/ephemerd/pkg/runtime" @@ -409,6 +410,13 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP if cfg.ModuleProxy.Enabled { gatewayPorts = append(gatewayPorts, modProxyPort) } + cargoProxyPort := cfg.CargoProxy.Port + if cargoProxyPort == 0 { + cargoProxyPort = 8083 + } + if cfg.CargoProxy.Enabled { + gatewayPorts = append(gatewayPorts, cargoProxyPort) + } // Initialize container networking net, err := networking.New(networking.Config{ @@ -436,16 +444,11 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP if upstream == "" { upstream = "https://proxy.golang.org" } - cleanup := cfg.ModuleProxy.Cleanup - if !cleanup { - cleanup = true - } - goProxy := goproxy.New(goproxy.Config{ CacheDir: joinPath(configDir, "cache", "gomod"), Upstream: upstream, ListenAddr: fmt.Sprintf("%s:%d", net.GatewayIP(), modProxyPort), - Cleanup: cleanup, + Cleanup: cfg.ModuleProxy.CleanupEnabled(), Log: log, }) if err := goProxy.Start(); err != nil { @@ -460,10 +463,43 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP } } - // Collect env vars from all cache proxies for injection into containers. + // Start Cargo/crates caching proxy if enabled + if cfg.CargoProxy.Enabled { + cargoProxy := cargoproxy.New(cargoproxy.Config{ + CacheDir: joinPath(configDir, "cache", "cargo"), + // Config dir sits OUTSIDE the cache so `ephemerd cache clear + // cargo` cannot delete the file mounted into running jobs. + ConfDir: joinPath(configDir, "cargo"), + IndexUpstream: cfg.CargoProxy.Upstream, + RustupUpstream: cfg.CargoProxy.RustupUpstream, + ListenAddr: fmt.Sprintf("%s:%d", net.GatewayIP(), cargoProxyPort), + IndexTTL: cfg.CargoProxy.IndexTTL, + Cleanup: cfg.CargoProxy.CleanupEnabled(), + Log: log, + }) + if err := cargoProxy.Start(); err != nil { + log.Warn("failed to start Cargo proxy, continuing without it", "error", err) + } else { + cacheProxies = append(cacheProxies, cargoProxy) + defer func() { + if err := cargoProxy.Stop(); err != nil { + log.Warn("error stopping Cargo proxy", "error", err) + } + }() + } + } + + // Collect env vars and mounts from all cache proxies for injection into + // containers. Only proxies that actually STARTED are in cacheProxies, so + // a failed proxy is never advertised to a job — that is the outer + // fail-open: jobs go straight to the upstream registry instead. var cacheProxyEnvVars []string + var cacheProxyMounts []proxies.Mount for _, cp := range cacheProxies { cacheProxyEnvVars = append(cacheProxyEnvVars, cp.EnvVars()...) + if mp, ok := cp.(proxies.MountProvider); ok { + cacheProxyMounts = append(cacheProxyMounts, mp.Mounts()...) + } } // Start the shared embedded BuildKit solver. One solver serves every @@ -516,6 +552,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP DindEnabled: cfg.Dind.Enabled, DindAllowPrivileged: cfg.Dind.ResolvedAllowPrivileged(), CacheProxyEnv: cacheProxyEnvVars, + CacheProxyMounts: cacheProxyMounts, Rlimits: cfg.Runtime.Rlimits.Resolved(), AllowNewPrivileges: cfg.Runtime.ResolvedAllowNewPrivileges(), Network: net, diff --git a/config.example.toml b/config.example.toml index 88cc680e..8661779b 100644 --- a/config.example.toml +++ b/config.example.toml @@ -203,6 +203,36 @@ shutdown_timeout = "5m" # [dispatch] # token = "..." +# Go module caching proxy. Runs a GOPROXY on the bridge gateway and injects +# GOPROXY into every job container, so `go mod download` hits the local cache +# instead of proxy.golang.org on each build. +# [module_proxy] +# enabled = false +# port = 8082 # listen port on the bridge gateway +# upstream = "https://proxy.golang.org" # fetched from on a cache miss +# cleanup = true # wipe the cache on shutdown + +# Cargo/crates caching proxy. A pull-through cache for the crates.io sparse +# index, .crate tarballs, and rustup toolchain artifacts. +# +# Jobs need no workflow changes: rustup is pointed at the proxy with +# RUSTUP_DIST_SERVER, and Cargo with a generated .cargo/config.toml that +# ephemerd bind-mounts read-only at the container's filesystem root. A repo +# that ships its own .cargo/config.toml still wins. +# +# Fails open: if the proxy does not start, nothing is injected and jobs go +# straight to crates.io. If an upstream is unreachable, cached data is served +# stale and uncached downloads are redirected to the origin. +# [cargo_proxy] +# enabled = false +# port = 8083 # listen port on the bridge gateway +# upstream = "https://index.crates.io" # sparse registry index +# rustup_upstream = "https://static.rust-lang.org" # toolchain distribution +# index_ttl = "10m" # index revalidation interval; +# # .crate tarballs are immutable +# # and cached permanently +# cleanup = false # keep the cache across restarts + [log] # Log level: debug, info, warn, error level = "info" diff --git a/docs/arch/sccache-evaluation.md b/docs/arch/sccache-evaluation.md new file mode 100644 index 00000000..8cb70e4e --- /dev/null +++ b/docs/arch/sccache-evaluation.md @@ -0,0 +1,197 @@ +# sccache for Rust compilation caching — evaluation + +**Status:** design note, no implementation. Written alongside the `[cargo_proxy]` +pull-through cache (`pkg/proxies/cargo`). + +**Bottom line:** worth building, but **after** the disk-pressure GC lands, and +only with a hard size cap wired into that GC from day one. A compile cache is +the single biggest available win on Rust CI wall-clock — and, if left +unbounded, the most likely next disk-fill incident. + +--- + +## Why consider it at all + +The `[cargo_proxy]` this note accompanies removes *download* cost: the sparse +index, `.crate` tarballs, and rustup toolchains stop crossing the network on +every build. That is a real saving, but it is not where Rust CI time goes. + +A cold Rust CI build spends a small minority of its wall-clock fetching crates +and the overwhelming majority in `rustc`. Dependencies are the bulk of it: a +project with a few hundred transitive crates compiles all of them before it +compiles a line of its own code, and it does so again on every fresh runner +because ephemerd's containers are ephemeral by design. `cargo build` alone +caches nothing across jobs — `target/` dies with the container. + +sccache addresses exactly that: it wraps `rustc`, hashes the compilation +inputs, and returns a cached object file on a hit. Unlike the registry proxy, +its benefit scales with dependency *compile* cost rather than dependency +*download* size, which is the dominant term. + +## Backend choice: local disk vs shared + +**Local disk (`SCCACHE_DIR`) on a host volume.** Simplest by far — a directory +on the runner host, bind-mounted into each job container. Hits come only from +jobs that previously ran *on this host*, which for a small fleet is most of +them. It has a built-in size cap (`SCCACHE_CACHE_SIZE`) with LRU eviction, +which is the single most important property given our history. No new network +service, no credentials, no cross-host trust boundary. + +**Shared/remote (S3, Redis, memcached, or `sccache --start-server` over +HTTP).** Hits are fleet-wide, so a cold host benefits from a warm one, and a +scale-out fleet converges on one cache instead of N. The costs are real +though: object storage or a Redis to operate, credentials to distribute into +job containers (a secret that untrusted CI code can read), and a write path +that lets any job poison the cache for every other host. It also reintroduces +network cost on the read path, partially undoing the thing we are optimising. + +**Recommendation: start local-disk-only.** It captures most of the benefit at a +fraction of the operational and security cost, and it is the configuration +whose eviction story we can actually enforce. Treat a shared backend as a +later, opt-in addition once there is evidence that cross-host misses matter. + +## Integration with ephemerd's job containers + +The mechanism is already built — `[cargo_proxy]` needed the same two levers, +and both generalise: + +- **Env vars** via `proxies.CacheProxy.EnvVars()`. sccache is configured + entirely through the environment: `RUSTC_WRAPPER=sccache`, `SCCACHE_DIR`, + `SCCACHE_CACHE_SIZE`. Unlike Cargo's source replacement, this needs no + config file, so no `MountProvider` is required for configuration. +- **A bind mount** via `proxies.MountProvider` for the cache directory itself + — but read-**write**, which is a materially different proposition from the + Cargo config mount (read-only). See the trust caveat below. + +The awkward part is the binary. `RUSTC_WRAPPER=sccache` requires an `sccache` +executable on `PATH` inside the container, and stock runner images do not have +one. Options, in increasing order of intrusiveness: + +1. Bind-mount a host-side `sccache` binary into the container and put its + directory on `PATH` (or set `RUSTC_WRAPPER` to the absolute mounted path, + avoiding the `PATH` edit entirely). Needs a statically-linked binary + matching the container's libc — musl builds exist and are the obvious + choice. This is the cheapest option and mirrors how the GitHub Actions + runner itself is already mounted in. +2. Ship it in ephemerd's embedded assets like the runner and CNI plugins, then + mount as above. Same runtime shape, adds a download to `mage download`. +3. Require the image to provide it. Rejected: it forces a workflow/image + change, which the `[cargo_proxy]` design explicitly avoided. + +Option 1 or 2. Note this makes the feature Linux-container-first; Windows and +the macOS native path would need separate handling and should be out of scope +for a first cut. + +## Correctness caveats + +sccache is conservative by design, but the failure mode of a compile cache is +*wrong output*, not a slow build, so this deserves care: + +- **Hash inputs.** sccache keys on the preprocessed source, the compiler + binary's hash, the full argument list, and the relevant env vars. That is + sound for ordinary `rustc` invocations. +- **Proc macros and build scripts are not cached.** `build.rs` output and + proc-macro expansion are outside sccache's model; crates that lean on them + see less benefit. This is a benefit ceiling, not a correctness risk. +- **Incremental compilation is incompatible.** sccache refuses to cache when + `CARGO_INCREMENTAL=1`. CI builds should set `CARGO_INCREMENTAL=0` anyway; + ephemerd should inject it alongside `RUSTC_WRAPPER` so the two settings can + never disagree. +- **Absolute paths leak into debug info.** Cached objects embed the paths they + were compiled from, so per-job workdir names (`/jobs//…`) + reduce the hit rate and can put a stale path in a backtrace. Mitigated with + `--remap-path-prefix`, which is worth injecting from the start rather than + retrofitting. +- **Toolchain churn invalidates everything.** A nightly bump changes the + compiler hash and orphans the entire cache. With a size cap and LRU this + self-heals; without one it silently doubles the footprint. +- **Trust boundary.** The cache directory is written by untrusted CI code and + read by the *next* job on the host. A malicious job can plant an object + under a key it predicts and have a later job link it. This is the same + category of risk the existing per-repo dind image cache manages by + namespacing per (provider, repo). The compile cache should do the same: + **partition the cache directory per repo**, not one shared pool. That + reduces the hit rate across repos and is the right trade. + +## Disk footprint — the part that must not be skipped + +We have just had two disk-exhaustion outages, both from caches that grew +without an eviction policy (most recently ~44 GB of BuildKit build cache in the +shared `buildkit` containerd namespace: 76 image records, 481 leases, 302 +snapshots, never cleaned). A compile cache is exactly the same shape of risk — +it is *designed* to accumulate — so it must not ship with the same gap. + +Non-negotiables for any implementation: + +- **A hard size cap, configured and enforced.** `SCCACHE_CACHE_SIZE` with + sccache's own LRU eviction, defaulting to something modest (10 GB is a + reasonable starting point) and surfaced as a config key, not a constant. + sccache enforces this itself, which makes it strictly better behaved than + BuildKit's cache has been. +- **Registration in `managedCaches()`** in `cmd/ephemerd/cache.go`, so + `ephemerd cache list` shows its size and `ephemerd cache clear sccache` + works. It is `LiveSafe: true` — a running job that loses a cache entry + simply recompiles. Follow the `cargo`/`gomod` entries as the template. +- **Participation in the disk-pressure GC** currently being built (which now + also covers the `buildkit` namespace). The GC must be able to evict from the + compile cache under pressure, and the compile cache must be low in the + eviction priority order — below anything that would force a re-download, and + well below live job state, but above nothing at all. A cache the GC cannot + see is a cache that will eventually fill the disk. +- **Per-repo partitioning interacts with the cap.** N repos × a per-repo cap is + the real ceiling. Either cap the aggregate and let the GC arbitrate, or size + the per-repo cap knowing the multiplier. Do not set a per-repo cap and quote + it as the total. + +## Interaction with BuildKit layer caching + +These two caches overlap and can double-count, which is worth designing around +rather than discovering later. + +BuildKit caches at *layer* granularity: a `docker build` whose Dockerfile runs +`cargo build` produces a layer keyed on the build context and the preceding +layers. sccache caches at *compilation unit* granularity, inside that build. +When both are active for the same work you can store the same compiled output +twice — once as object files in the sccache dir, once inside a BuildKit +snapshot — and the BuildKit copy is the one that has already caused an +incident. + +Two clean positions: + +- **Jobs that compile Rust directly on the runner** (`cargo build` as a + workflow step, which is the ephpm case): sccache applies, BuildKit is not + involved. No overlap. This is the case worth optimising for. +- **Jobs that compile Rust inside `docker build`**: BuildKit's layer cache + already covers a clean rebuild, and threading sccache into the builder means + plumbing the cache into the build (a cache mount) plus the env vars. The + incremental benefit over a working BuildKit cache is small. + +**Recommendation: scope sccache to the direct-`cargo build` path and do not +wire it into `docker build`.** That keeps the two caches disjoint by +construction, avoids two overlapping unbounded stores, and targets the case +that actually dominates. If BuildKit-side Rust caching is wanted later, the +right lever is a BuildKit cache mount, not a second copy of sccache. + +## Recommendation + +Build it, with these conditions: + +1. **Sequence it after the disk-pressure GC.** The GC is the prerequisite, not + a follow-up. Shipping an unbounded compile cache into a fleet that has just + had two disk outages would be repeating the mistake. +2. **Local disk backend, per-repo partitioned, hard size cap, registered in + `managedCaches()` and in the GC's eviction order.** +3. **Inject `RUSTC_WRAPPER`, `SCCACHE_DIR`, `SCCACHE_CACHE_SIZE`, + `CARGO_INCREMENTAL=0`, and `--remap-path-prefix` together**, via the + existing `CacheProxy`/`MountProvider` plumbing. Reuse it; do not invent a + parallel mechanism. +4. **Scope to the direct `cargo build` path**, Linux containers first. +5. **Fail open**, exactly as `[cargo_proxy]` does: if the binary or the cache + dir is unavailable, inject nothing and let the build compile normally. + sccache's own `SCCACHE_ERROR_LOG` should be surfaced, but a cache failure + must never fail a job. + +Expected payoff is a large reduction in Rust CI wall-clock on warm hosts — +substantially bigger than the registry proxy's, because it removes compile +time rather than download time. The risk is entirely in the disk dimension, +and it is fully mitigable with a cap the tool already implements. diff --git a/docs/cli/cache.md b/docs/cli/cache.md index 5e8d1328..a48759af 100644 --- a/docs/cli/cache.md +++ b/docs/cli/cache.md @@ -20,6 +20,7 @@ ephemerd cache clear --all [--yes] |------|-----------------------------|-------------|-------------------------| | `images` | `images/` | Staged OCI image tarballs imported into containerd on startup | yes | | `gomod` | `cache/gomod/` | Go module proxy cache (GOPROXY) served to job containers | yes | +| `cargo` | `cache/cargo/` | Cargo proxy cache (crates.io sparse index, `.crate` tarballs, rustup toolchains) served to job containers | yes | | `buildkit` | `buildkit/` | Embedded BuildKit solver cache + history (`docker build` layers) | no | | `worker` | `worker/` | BuildKit worker snapshot/content root | no | | `runners` | `runners/` | Extracted GitHub Actions runner binaries (per version/OS) | no | diff --git a/docs/getting-started/configuration.md b/docs/getting-started/configuration.md index 630c0da1..d4bcd4f2 100644 --- a/docs/getting-started/configuration.md +++ b/docs/getting-started/configuration.md @@ -117,6 +117,21 @@ max_concurrent = 4 # max simultaneous jobs # cache_prune_interval = "24h" # how often the per-repo image cache pruner runs # cache_max_age = "168h" # evict cached image records inactive longer than this (7 days) +# --- Package caching proxies -------------------------------------------------- +# [module_proxy] +# enabled = false # run a GOPROXY on the bridge gateway +# port = 8082 # listen port +# upstream = "https://proxy.golang.org" +# cleanup = true # wipe the cache on shutdown + +# [cargo_proxy] +# enabled = false # pull-through cache for crates.io + rustup +# port = 8083 # listen port +# upstream = "https://index.crates.io" +# rustup_upstream = "https://static.rust-lang.org" +# index_ttl = "10m" # sparse-index revalidation interval +# cleanup = false # keep the cache across restarts + # --- Metrics ------------------------------------------------------------------ [metrics] # enabled = false # expose Prometheus /metrics endpoint @@ -284,6 +299,57 @@ The cache namespace persists across jobs and across ephemerd restarts. Per-job s **Disabling caching.** Setting `cache_max_age = "0"` disables eviction (the cache grows unbounded — useful for debugging but not recommended in production). Setting `cache_prune_interval = "0"` disables the pruner goroutine entirely; equivalent to "keep everything forever, even empty namespaces." +### `[module_proxy]` + +Go module caching proxy. ephemerd runs a single GOPROXY on the bridge gateway and injects `GOPROXY=http://:|direct` into every job container, so repeated `go mod download` runs hit the local disk cache instead of `proxy.golang.org`. + +| Field | Type | Default | Description | +|---|---|---|---| +| `enabled` | boolean | `false` | Run the Go module proxy | +| `port` | integer | `8082` | Listen port on the bridge gateway | +| `upstream` | string | `"https://proxy.golang.org"` | Fetched from on a cache miss | +| `cleanup` | boolean | `true` | Wipe the cache directory on shutdown. Set `false` to keep it across restarts. | + +Immutable module files (`.info`, `.mod`, `.zip`) are cached; mutable endpoints (`@latest`, `@v/list`) and `sumdb` requests pass through. The `|direct` separator means the go command falls back to the origin on **any** proxy error, so a broken cache slows a build rather than failing it. + +### `[cargo_proxy]` + +Cargo/crates caching proxy — the Rust counterpart to `[module_proxy]`. One HTTP server on the bridge gateway serves three routes: + +| Route | Upstream | Caching | +|---|---|---| +| `/index/…` | `upstream` (sparse registry index) | **Mutable** — served from cache for `index_ttl`, then revalidated with a conditional GET | +| `/crates/{name}/{version}/download` | the registry's own `dl` template | **Immutable** — cached permanently, never refetched | +| `/rustup/…` | `rustup_upstream` | Dated artifacts (`dist/YYYY-MM-DD/…`) immutable; channel manifests revalidated on `index_ttl` | + +| Field | Type | Default | Description | +|---|---|---|---| +| `enabled` | boolean | `false` | Run the Cargo proxy | +| `port` | integer | `8083` | Listen port on the bridge gateway | +| `upstream` | string | `"https://index.crates.io"` | Sparse registry index | +| `rustup_upstream` | string | `"https://static.rust-lang.org"` | Toolchain distribution server | +| `index_ttl` | duration | `"10m"` | How long a cached index entry is served before revalidation. A negative value revalidates on every request. | +| `cleanup` | boolean | `false` | Wipe the cache on shutdown. Defaults to **false**, unlike `[module_proxy]` — a pull-through cache that empties itself on every restart saves nothing. | + +**How jobs pick it up — no workflow changes required.** + +- **rustup** reads its mirror from the environment, so ephemerd injects `RUSTUP_DIST_SERVER`. +- **Cargo does not.** Source replacement (`[source.crates-io] replace-with`) is the only mechanism Cargo offers for redirecting crates.io, and it is read **exclusively from config files** — `CARGO_SOURCE_*` environment variables are silently ignored. ephemerd therefore generates a `.cargo/config.toml` under `/cargo/` and bind-mounts it **read-only** at the container's filesystem root (`/.cargo`, or `C:\.cargo` on Windows). Cargo searches the current directory and *every ancestor* for `.cargo/config.toml`, so a file at the root applies to any workspace path a job checks out — no knowledge of the checkout location, the job user's home, or the image's `CARGO_HOME` is needed, and `CARGO_HOME` itself is left untouched so it stays writable. + +A repository that ships its own `.cargo/config.toml` still wins: Cargo prefers config closer to the workspace. + +**Fail-open behaviour.** A cache must never turn a registry hiccup into a red CI job, so failures degrade in three layers: + +1. If the proxy does not start, nothing is injected and no mount is added — jobs go straight to crates.io. +2. If an upstream is unreachable or returns 5xx and a cached copy exists, the **stale copy is served** with a warning. +3. If nothing is cached, crate and rustup downloads are answered with a `307` redirect to the real origin, so the job fetches it directly (slower, uncached) instead of failing. + +Genuine `404`s are passed through as `404` — a nonexistent crate version must stay distinguishable from an outage. + +**Scope.** The mount lands in the runner container. Jobs that run their steps inside a *further* container (a `container:` image spawned via Docker-in-Docker) do not inherit it. + +**Cache location.** Cached content lives at `/cache/cargo/` and is visible to `ephemerd cache list` / `ephemerd cache clear cargo`. The generated container config lives at `/cargo/` — deliberately outside the cache root, so clearing the cache cannot pull the mounted config out from under a running job. + ### `[metrics]` Prometheus metrics endpoint. diff --git a/pkg/config/config.go b/pkg/config/config.go index 6381722b..ea80bc9e 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -28,6 +28,7 @@ type Config struct { VM VMConfig `toml:"vm"` Dind DindConfig `toml:"dind"` ModuleProxy ModuleProxyConfig `toml:"module_proxy"` + CargoProxy CargoProxyConfig `toml:"cargo_proxy"` Runtime RuntimeConfig `toml:"runtime"` Runner RunnerConfig `toml:"runner"` Metrics MetricsConfig `toml:"metrics"` @@ -380,7 +381,59 @@ type ModuleProxyConfig struct { Enabled bool `toml:"enabled"` // enable Go module caching proxy Port int `toml:"port"` // listen port on bridge gateway (default 8082) Upstream string `toml:"upstream"` // upstream proxy URL (default "https://proxy.golang.org") - Cleanup bool `toml:"cleanup"` // wipe cache on shutdown (default true) + Cleanup *bool `toml:"cleanup"` // wipe cache on shutdown (default true) +} + +// CleanupEnabled reports whether the Go module cache is wiped on shutdown. +// Defaults to true, preserving the historical behavior. +// +// Pointer-typed so "unset" is distinguishable from an explicit false. The +// previous plain bool could not express that: the call site coerced any +// false value back to true, which silently ignored `cleanup = false`. +func (m *ModuleProxyConfig) CleanupEnabled() bool { + if m.Cleanup == nil { + return true + } + return *m.Cleanup +} + +// CargoProxyConfig configures the Cargo/crates caching proxy. +// +// When enabled, ephemerd runs a pull-through cache for the crates.io sparse +// index, .crate tarballs, and rustup toolchain artifacts on the bridge +// gateway. Job containers are pointed at it automatically: rustup via +// RUSTUP_DIST_SERVER, and Cargo via a generated .cargo/config.toml that is +// bind-mounted read-only at the container's filesystem root (Cargo ignores +// CARGO_SOURCE_* environment variables, so a file is the only mechanism). +type CargoProxyConfig struct { + // Enabled turns the proxy on. Default false. + Enabled bool `toml:"enabled"` + // Port is the listen port on the bridge gateway. Default 8083. + Port int `toml:"port"` + // Upstream is the sparse registry index base URL. + // Default "https://index.crates.io". + Upstream string `toml:"upstream"` + // RustupUpstream is the toolchain distribution server. + // Default "https://static.rust-lang.org". + RustupUpstream string `toml:"rustup_upstream"` + // IndexTTL is how long a cached sparse-index entry is served before a + // conditional revalidation. Default 10m. Crate tarballs ignore this — + // they are immutable and cached permanently. + IndexTTL time.Duration `toml:"index_ttl"` + // Cleanup wipes the cache on shutdown. Default FALSE, unlike the Go + // module proxy: the whole point of a pull-through cache is to survive + // restarts, and wiping it on every shutdown is what made the module + // proxy's cache worthless. + Cleanup *bool `toml:"cleanup"` +} + +// CleanupEnabled reports whether the Cargo cache is wiped on shutdown. +// Defaults to false — see the field comment. +func (c *CargoProxyConfig) CleanupEnabled() bool { + if c.Cleanup == nil { + return false + } + return *c.Cleanup } // VMConfig configures virtual machines for cross-OS job execution. diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index ccdbe1d9..d1b3a691 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -908,8 +908,78 @@ cleanup = false if cfg.ModuleProxy.Upstream != "https://goproxy.io" { t.Errorf("ModuleProxy.Upstream = %q, want %q", cfg.ModuleProxy.Upstream, "https://goproxy.io") } - if cfg.ModuleProxy.Cleanup { - t.Error("ModuleProxy.Cleanup = true, want false") + if cfg.ModuleProxy.CleanupEnabled() { + t.Error("ModuleProxy.CleanupEnabled() = true, want false (cleanup = false was set explicitly)") + } +} + +// TestModuleProxyCleanupDefault pins that an unset cleanup still defaults to +// true — the pointer field must not change existing behavior, only make an +// explicit false actually take effect. +func TestModuleProxyCleanupDefault(t *testing.T) { + var m ModuleProxyConfig + if !m.CleanupEnabled() { + t.Error("ModuleProxy.CleanupEnabled() = false when unset, want true") + } +} + +// --- CargoProxy config --- + +func TestLoad_CargoProxyConfig(t *testing.T) { + t.Setenv("GITHUB_TOKEN", "") + tmp := t.TempDir() + path := filepath.Join(tmp, "config.toml") + if err := os.WriteFile(path, []byte(` +[github] +token = "ghp_test" +owner = "org" + +[cargo_proxy] +enabled = true +port = 9100 +upstream = "https://index.example.test" +rustup_upstream = "https://static.example.test" +index_ttl = "5m" +cleanup = true +`), 0644); err != nil { + t.Fatal(err) + } + + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load() error: %v", err) + } + + if !cfg.CargoProxy.Enabled { + t.Error("CargoProxy.Enabled = false, want true") + } + if cfg.CargoProxy.Port != 9100 { + t.Errorf("CargoProxy.Port = %d, want 9100", cfg.CargoProxy.Port) + } + if cfg.CargoProxy.Upstream != "https://index.example.test" { + t.Errorf("CargoProxy.Upstream = %q", cfg.CargoProxy.Upstream) + } + if cfg.CargoProxy.RustupUpstream != "https://static.example.test" { + t.Errorf("CargoProxy.RustupUpstream = %q", cfg.CargoProxy.RustupUpstream) + } + if cfg.CargoProxy.IndexTTL != 5*time.Minute { + t.Errorf("CargoProxy.IndexTTL = %v, want 5m", cfg.CargoProxy.IndexTTL) + } + if !cfg.CargoProxy.CleanupEnabled() { + t.Error("CargoProxy.CleanupEnabled() = false, want true") + } +} + +// TestCargoProxyDefaults pins the defaults that matter: the proxy is opt-in, +// and — unlike the Go module proxy — its cache is NOT wiped on shutdown. A +// pull-through cache that empties itself on every restart saves nothing. +func TestCargoProxyDefaults(t *testing.T) { + var c CargoProxyConfig + if c.Enabled { + t.Error("CargoProxy.Enabled = true when unset, want false (opt-in)") + } + if c.CleanupEnabled() { + t.Error("CargoProxy.CleanupEnabled() = true when unset, want false") } } @@ -2397,12 +2467,12 @@ func TestResolvedReconcileInterval(t *testing.T) { in string want time.Duration }{ - {"", 30 * time.Minute}, // default (backstop only; event-driven self-heal is primary) - {"2m", 2 * time.Minute}, // custom - {"90s", 90 * time.Second}, // custom - {"garbage", 30 * time.Minute}, // unparseable -> default - {"0s", 0}, // explicit disable - {"-1m", 0}, // negative -> disabled + {"", 30 * time.Minute}, // default (backstop only; event-driven self-heal is primary) + {"2m", 2 * time.Minute}, // custom + {"90s", 90 * time.Second}, // custom + {"garbage", 30 * time.Minute}, // unparseable -> default + {"0s", 0}, // explicit disable + {"-1m", 0}, // negative -> disabled } for _, c := range cases { w := &WebhookConfig{ReconcileInterval: c.in} diff --git a/pkg/proxies/cargo/cache.go b/pkg/proxies/cargo/cache.go new file mode 100644 index 00000000..a532bc77 --- /dev/null +++ b/pkg/proxies/cargo/cache.go @@ -0,0 +1,355 @@ +package cargoproxy + +import ( + "encoding/json" + "fmt" + "path" + "path/filepath" + "regexp" + "strings" + "time" +) + +// freshness is the decision the cache makes for one request. Kept as a pure +// value so the TTL/revalidation policy can be table-tested without a server, +// a clock, or a filesystem. +type freshness int + +const ( + // fetchFresh: nothing usable on disk — fetch unconditionally. + fetchFresh freshness = iota + // serveCached: the cached copy is within its TTL (or immutable) — serve + // it and do not touch the network at all. + serveCached + // revalidate: a cached copy exists but has aged out — issue a + // conditional GET (If-None-Match / If-Modified-Since) so the common + // "nothing changed" case costs one 304 instead of a full body. + revalidate +) + +func (f freshness) String() string { + switch f { + case fetchFresh: + return "fetch" + case serveCached: + return "hit" + case revalidate: + return "revalidate" + default: + return "unknown" + } +} + +// entryMeta is the sidecar record stored next to every cached body. It holds +// the upstream validators so a stale entry can be revalidated cheaply. +type entryMeta struct { + ETag string `json:"etag,omitempty"` + LastModified string `json:"last_modified,omitempty"` + ContentType string `json:"content_type,omitempty"` + Fetched time.Time `json:"fetched"` +} + +// decideFreshness is the whole cache policy in one pure function. +// +// - immutable content (.crate tarballs, dated rustup artifacts) is +// content-addressed by its URL: once cached it is NEVER refetched and +// never revalidated, whatever its age. +// - mutable content (the sparse index, rustup channel manifests) is served +// from cache inside ttl and conditionally revalidated after that, so we +// never serve an indefinitely stale index. +// +// A zero or negative ttl means "always revalidate", which is the safe +// direction for mutable data. +func decideFreshness(cached bool, immutable bool, fetched, now time.Time, ttl time.Duration) freshness { + if !cached { + return fetchFresh + } + if immutable { + return serveCached + } + if ttl > 0 && now.Sub(fetched) < ttl { + return serveCached + } + return revalidate +} + +// crateNameRe matches the crate names crates.io permits: ASCII alphanumerics +// plus '-' and '_'. Anything else is rejected rather than sanitized — a name +// we do not recognise must never reach the filesystem. +var crateNameRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +// crateVersionRe is deliberately broader than strict semver (pre-release and +// build metadata are common) but still excludes separators and traversal. +var crateVersionRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._+-]{0,63}$`) + +// safeSegments splits a URL path into segments and rejects anything that +// could escape the cache root or confuse the host filesystem: traversal, +// empty segments, absolute/UNC forms, backslashes, drive letters, control +// characters, and Windows-hostile names. This is the single choke point every +// on-disk cache path goes through. +func safeSegments(urlPath string) ([]string, error) { + clean := strings.Trim(urlPath, "/") + if clean == "" { + return nil, fmt.Errorf("empty path") + } + segs := strings.Split(clean, "/") + out := make([]string, 0, len(segs)) + for _, s := range segs { + if s == "" { + return nil, fmt.Errorf("empty path segment in %q", urlPath) + } + if s == "." || s == ".." { + return nil, fmt.Errorf("path traversal segment %q in %q", s, urlPath) + } + if strings.ContainsAny(s, `\:*?"<>|`) { + return nil, fmt.Errorf("illegal character in path segment %q", s) + } + for _, r := range s { + if r < 0x20 || r == 0x7f { + return nil, fmt.Errorf("control character in path segment %q", s) + } + } + out = append(out, s) + } + return out, nil +} + +// cachePathUnder joins validated URL-path segments onto a cache root. +func cachePathUnder(root, urlPath string) (string, error) { + segs, err := safeSegments(urlPath) + if err != nil { + return "", err + } + return filepath.Join(append([]string{root}, segs...)...), nil +} + +// indexCachePath maps a sparse-index request path (e.g. "/se/rd/serde") to +// its on-disk location under /index. +func indexCachePath(cacheDir, indexPath string) (string, error) { + p, err := cachePathUnder(filepath.Join(cacheDir, "index"), indexPath) + if err != nil { + return "", fmt.Errorf("index cache path for %q: %w", indexPath, err) + } + return p, nil +} + +// crateCachePath maps a crate name+version to its on-disk .crate location. +// Both components are validated against the registry's own character rules, +// so a hostile "name" like "../../etc/passwd" is rejected outright rather +// than escaping the cache root. +func crateCachePath(cacheDir, name, version string) (string, error) { + if !crateNameRe.MatchString(name) { + return "", fmt.Errorf("invalid crate name %q", name) + } + if !crateVersionRe.MatchString(version) { + return "", fmt.Errorf("invalid crate version %q for %q", version, name) + } + lower := strings.ToLower(name) + return filepath.Join(cacheDir, "crates", lower, lower+"-"+version+".crate"), nil +} + +// rustupCachePath maps a rustup dist path to its on-disk location. +func rustupCachePath(cacheDir, distPath string) (string, error) { + p, err := cachePathUnder(filepath.Join(cacheDir, "rustup"), distPath) + if err != nil { + return "", fmt.Errorf("rustup cache path for %q: %w", distPath, err) + } + return p, nil +} + +// metaPath is the sidecar path for a cached body. Crate names and index +// entries never contain a dot, and rustup artifacts are immutable (no +// sidecar), so ".meta" cannot collide with a real cache entry. +func metaPath(bodyPath string) string { return bodyPath + ".meta" } + +// datedRustupRe matches the YYYY-MM-DD segment that rustup uses for pinned, +// immutable nightly/dated artifacts. +var datedRustupRe = regexp.MustCompile(`(^|/)\d{4}-\d{2}-\d{2}(/|$)`) + +// isImmutableRustupPath reports whether a rustup dist path is content-stable. +// +// Dated artifacts (dist/2026-08-01/rust-std-nightly-x86_64...tar.xz) are +// published once and never rewritten, so they cache forever. The channel +// manifests (dist/channel-rust-nightly.toml and its .sha256/.asc siblings) +// are rewritten daily in place, so they must revalidate — caching those +// forever would pin every job to the toolchain that happened to be current +// when the daemon first started. +func isImmutableRustupPath(distPath string) bool { + return datedRustupRe.MatchString(distPath) +} + +// isImmutableIndexPath reports whether a sparse-index path is immutable. +// Nothing in the index is: entries gain a line whenever a version is +// published or yanked, and config.json can be re-pointed. Kept as a named +// function so the call sites read symmetrically and the intent is explicit +// rather than an unexplained "false". +func isImmutableIndexPath(string) bool { return false } + +// indexPrefix returns the sparse-index directory prefix for a crate name, +// per the registry index layout: 1-char names live under "1", 2-char under +// "2", 3-char under "3/", and everything else under the first +// two characters then the next two. +func indexPrefix(name string) string { + switch len(name) { + case 0: + return "" + case 1: + return "1" + case 2: + return "2" + case 3: + return path.Join("3", name[0:1]) + default: + return path.Join(name[0:2], name[2:4]) + } +} + +// defaultDLTemplate is the download URL used when the upstream registry's +// config.json is not available (first crate request after a restart with an +// empty index cache). It is crates.io's published template. +const defaultDLTemplate = "https://static.crates.io/crates/{crate}/{crate}-{version}.crate" + +// registryConfig is the subset of a sparse registry's config.json we care +// about. Everything else is preserved verbatim on rewrite via rawConfig. +type registryConfig struct { + DL string `json:"dl"` + API string `json:"api,omitempty"` +} + +// parseDLTemplate extracts the "dl" template from a registry config.json. +// A config.json we cannot parse falls back to the crates.io default rather +// than failing the request — the worst case is that we proxy from the +// canonical CDN instead of a mirror. +func parseDLTemplate(raw []byte) string { + var cfg registryConfig + if err := json.Unmarshal(raw, &cfg); err != nil || strings.TrimSpace(cfg.DL) == "" { + return defaultDLTemplate + } + return cfg.DL +} + +// dlMarkers are the substitutions a registry may use in its "dl" template. +var dlMarkers = []string{"{crate}", "{version}", "{prefix}", "{lowerprefix}", "{sha256-checksum}"} + +// expandDL turns a registry "dl" template into a concrete upstream URL for +// one crate version, implementing the registry-index rules: +// +// - {crate}, {version}, {prefix}, {lowerprefix} are substituted; +// - a template containing NO markers gets "/{crate}/{version}/download" +// appended, which is the documented default; +// - {sha256-checksum} cannot be expanded here (the checksum lives in the +// index entry, which we do not parse), so such a template is reported +// unusable and the caller falls back to the crates.io default. +func expandDL(tmpl, name, version string) (string, bool) { + tmpl = strings.TrimSpace(tmpl) + if tmpl == "" { + return "", false + } + if strings.Contains(tmpl, "{sha256-checksum}") { + return "", false + } + + hasMarker := false + for _, m := range dlMarkers { + if strings.Contains(tmpl, m) { + hasMarker = true + break + } + } + if !hasMarker { + return strings.TrimRight(tmpl, "/") + "/" + name + "/" + version + "/download", true + } + + lower := strings.ToLower(name) + r := strings.NewReplacer( + "{crate}", name, + "{version}", version, + "{prefix}", indexPrefix(name), + "{lowerprefix}", indexPrefix(lower), + ) + return r.Replace(tmpl), true +} + +// rewriteConfigJSON rewrites a registry config.json so Cargo fetches crate +// tarballs from THIS proxy instead of the upstream CDN, while leaving every +// other field — notably "api", which `cargo publish`/`cargo search` use — +// pointing at the real registry. The rewritten "dl" is deliberately +// marker-free, so Cargo appends "/{crate}/{version}/download" and we get a +// single, unambiguous request shape to route. +// +// Unknown fields are preserved: the config is decoded into a generic map so +// a future registry key is passed through rather than silently dropped. +func rewriteConfigJSON(raw []byte, base string) ([]byte, error) { + var generic map[string]any + if err := json.Unmarshal(raw, &generic); err != nil { + return nil, fmt.Errorf("parsing registry config.json: %w", err) + } + generic["dl"] = strings.TrimRight(base, "/") + cratesRoute + out, err := json.Marshal(generic) + if err != nil { + return nil, fmt.Errorf("re-encoding registry config.json: %w", err) + } + return out, nil +} + +// parseCrateDownloadPath splits the crate-download route Cargo generates +// from our rewritten "dl" — "/crates///download" — into its +// components. Any other shape is rejected. +func parseCrateDownloadPath(urlPath string) (name, version string, ok bool) { + rest := strings.TrimPrefix(urlPath, cratesRoute+"/") + if rest == urlPath { + return "", "", false + } + parts := strings.Split(rest, "/") + if len(parts) != 3 || parts[2] != "download" { + return "", "", false + } + if !crateNameRe.MatchString(parts[0]) || !crateVersionRe.MatchString(parts[1]) { + return "", "", false + } + return parts[0], parts[1], true +} + +// containerConfigTOML renders the Cargo config that gets mounted into every +// job container. +// +// Source replacement is the ONLY mechanism Cargo offers for redirecting +// crates.io, and it is read exclusively from config files — CARGO_SOURCE_* +// environment variables are silently ignored (verified against cargo stable: +// with the env vars set, cargo still reported "Updating crates.io index"). +// Hence a generated file rather than an env var. +func containerConfigTOML(base string) string { + base = strings.TrimRight(base, "/") + return `# Generated by ephemerd. Do NOT edit — regenerated on every daemon start. +# +# Routes crates.io through ephemerd's on-host pull-through cache so repeated +# CI builds do not re-download the same .crate tarballs. Delete the +# [cargo_proxy] block from ephemerd's config.toml to turn this off. +# +# A repository's own .cargo/config.toml still wins: Cargo prefers config +# closer to the workspace, so a project that pins its own mirror is +# unaffected by this file. + +[source.crates-io] +replace-with = "ephemerd" + +[source.ephemerd] +registry = "sparse+` + base + indexRoute + `/" + +[net] +retry = 3 +` +} + +// containerConfigDest is the container path the generated .cargo directory is +// mounted at. Cargo walks the current directory and EVERY ancestor looking +// for .cargo/config.toml, so placing it at the filesystem root makes it apply +// to any workspace path a job might use — no knowledge of the checkout +// location, the job user's home, or the image's CARGO_HOME required, and +// CARGO_HOME itself is left alone so it stays writable. +func containerConfigDest(goos string) string { + if goos == "windows" { + return `C:\.cargo` + } + return "/.cargo" +} diff --git a/pkg/proxies/cargo/cache_test.go b/pkg/proxies/cargo/cache_test.go new file mode 100644 index 00000000..9c00de4e --- /dev/null +++ b/pkg/proxies/cargo/cache_test.go @@ -0,0 +1,542 @@ +package cargoproxy + +import ( + "encoding/json" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestDecideFreshness(t *testing.T) { + now := time.Date(2026, 8, 11, 12, 0, 0, 0, time.UTC) + ttl := 10 * time.Minute + + tests := []struct { + name string + cached bool + immutable bool + fetched time.Time + ttl time.Duration + want freshness + }{ + { + name: "nothing cached fetches", + want: fetchFresh, + }, + { + name: "nothing cached fetches even when immutable", + immutable: true, + want: fetchFresh, + }, + { + name: "immutable is served regardless of age", + cached: true, + immutable: true, + fetched: now.Add(-5000 * time.Hour), + ttl: ttl, + want: serveCached, + }, + { + name: "immutable with zero fetch time is still served", + cached: true, + immutable: true, + ttl: ttl, + want: serveCached, + }, + { + name: "mutable inside ttl is served", + cached: true, + fetched: now.Add(-1 * time.Minute), + ttl: ttl, + want: serveCached, + }, + { + name: "mutable exactly at ttl revalidates", + cached: true, + fetched: now.Add(-ttl), + ttl: ttl, + want: revalidate, + }, + { + name: "mutable past ttl revalidates", + cached: true, + fetched: now.Add(-1 * time.Hour), + ttl: ttl, + want: revalidate, + }, + { + name: "zero ttl always revalidates", + cached: true, + fetched: now, + ttl: 0, + want: revalidate, + }, + { + name: "negative ttl always revalidates", + cached: true, + fetched: now, + ttl: -time.Minute, + want: revalidate, + }, + { + name: "clock skew (future fetch time) still serves", + cached: true, + fetched: now.Add(time.Hour), + ttl: ttl, + want: serveCached, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := decideFreshness(tt.cached, tt.immutable, tt.fetched, now, tt.ttl) + if got != tt.want { + t.Errorf("decideFreshness(cached=%v, immutable=%v, age=%v, ttl=%v) = %v, want %v", + tt.cached, tt.immutable, now.Sub(tt.fetched), tt.ttl, got, tt.want) + } + }) + } +} + +func TestIndexPrefix(t *testing.T) { + tests := []struct{ name, want string }{ + {"a", "1"}, + {"go", "2"}, + {"bar", "3/b"}, + {"serde", "se/rd"}, + {"tokio", "to/ki"}, + {"cc", "2"}, + {"", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := indexPrefix(tt.name); got != tt.want { + t.Errorf("indexPrefix(%q) = %q, want %q", tt.name, got, tt.want) + } + }) + } +} + +func TestSafeSegments_RejectsTraversalAndJunk(t *testing.T) { + bad := []string{ + "", + "/", + "//", + "..", + "/../etc/passwd", + "/se/../../etc/passwd", + "/se/rd/../../../root", + "/./serde", + `/se\rd\serde`, + "/C:/windows", + "/se//rd", + "/se/rd/ser\x00de", + "/se/rd/ser\nde", + "/se/rd/ser|de", + "/se/rd/ser*de", + } + for _, p := range bad { + t.Run(strings.ReplaceAll(p, "\x00", "NUL"), func(t *testing.T) { + if segs, err := safeSegments(p); err == nil { + t.Errorf("safeSegments(%q) = %v, want error", p, segs) + } + }) + } + + good := map[string][]string{ + "/se/rd/serde": {"se", "rd", "serde"}, + "se/rd/serde": {"se", "rd", "serde"}, + "/1/a": {"1", "a"}, + "/3/b/bar": {"3", "b", "bar"}, + "/config.json": {"config.json"}, + "/dist/x.tar.z": {"dist", "x.tar.z"}, + } + for p, want := range good { + t.Run("ok"+p, func(t *testing.T) { + got, err := safeSegments(p) + if err != nil { + t.Fatalf("safeSegments(%q): %v", p, err) + } + if strings.Join(got, "/") != strings.Join(want, "/") { + t.Errorf("safeSegments(%q) = %v, want %v", p, got, want) + } + }) + } +} + +// TestCachePaths_StayUnderRoot is the security property: no request path may +// produce a cache location outside the cache root. +func TestCachePaths_StayUnderRoot(t *testing.T) { + root := filepath.Join("cachedir") + + t.Run("index", func(t *testing.T) { + got, err := indexCachePath(root, "/se/rd/serde") + if err != nil { + t.Fatalf("indexCachePath: %v", err) + } + want := filepath.Join(root, "index", "se", "rd", "serde") + if got != want { + t.Errorf("indexCachePath = %q, want %q", got, want) + } + if _, err := indexCachePath(root, "/../../escape"); err == nil { + t.Error("indexCachePath accepted a traversal path") + } + }) + + t.Run("rustup", func(t *testing.T) { + got, err := rustupCachePath(root, "/dist/2026-08-01/rust-std.tar.xz") + if err != nil { + t.Fatalf("rustupCachePath: %v", err) + } + want := filepath.Join(root, "rustup", "dist", "2026-08-01", "rust-std.tar.xz") + if got != want { + t.Errorf("rustupCachePath = %q, want %q", got, want) + } + if _, err := rustupCachePath(root, "/dist/../../escape"); err == nil { + t.Error("rustupCachePath accepted a traversal path") + } + }) +} + +func TestCrateCachePath(t *testing.T) { + root := "cachedir" + tests := []struct { + name, crate, version, want string + wantErr bool + }{ + {name: "simple", crate: "serde", version: "1.0.203", want: filepath.Join(root, "crates", "serde", "serde-1.0.203.crate")}, + {name: "underscores and dashes", crate: "proc-macro2", version: "1.0.86", want: filepath.Join(root, "crates", "proc-macro2", "proc-macro2-1.0.86.crate")}, + {name: "case normalised", crate: "Inflector", version: "0.11.4", want: filepath.Join(root, "crates", "inflector", "inflector-0.11.4.crate")}, + {name: "prerelease version", crate: "tokio", version: "1.0.0-beta.1", want: filepath.Join(root, "crates", "tokio", "tokio-1.0.0-beta.1.crate")}, + {name: "build metadata", crate: "tokio", version: "1.0.0+build.5", want: filepath.Join(root, "crates", "tokio", "tokio-1.0.0+build.5.crate")}, + + {name: "traversal in name", crate: "../../etc/passwd", version: "1.0.0", wantErr: true}, + {name: "slash in name", crate: "a/b", version: "1.0.0", wantErr: true}, + {name: "dot in name", crate: "..", version: "1.0.0", wantErr: true}, + {name: "empty name", crate: "", version: "1.0.0", wantErr: true}, + {name: "traversal in version", crate: "serde", version: "../../x", wantErr: true}, + {name: "empty version", crate: "serde", version: "", wantErr: true}, + {name: "backslash in version", crate: "serde", version: `1.0\..\x`, wantErr: true}, + {name: "version starting with dot", crate: "serde", version: ".hidden", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := crateCachePath(root, tt.crate, tt.version) + if tt.wantErr { + if err == nil { + t.Fatalf("crateCachePath(%q, %q) = %q, want error", tt.crate, tt.version, got) + } + return + } + if err != nil { + t.Fatalf("crateCachePath(%q, %q): %v", tt.crate, tt.version, err) + } + if got != tt.want { + t.Errorf("crateCachePath = %q, want %q", got, tt.want) + } + }) + } +} + +func TestIsImmutableRustupPath(t *testing.T) { + tests := []struct { + path string + want bool + }{ + // Dated artifacts are published once and never rewritten. + {"/dist/2026-08-01/rust-std-nightly-x86_64-unknown-linux-gnu.tar.xz", true}, + {"/dist/2026-08-01/channel-rust-nightly.toml", true}, + {"/dist/2026-08-01/channel-rust-nightly.toml.sha256", true}, + {"dist/2024-12-31/rustc-nightly-src.tar.gz", true}, + {"/dist/2026-08-01", true}, + + // Channel manifests are rewritten in place — must revalidate. + {"/dist/channel-rust-nightly.toml", false}, + {"/dist/channel-rust-stable.toml", false}, + {"/dist/channel-rust-beta.toml.sha256", false}, + {"/rustup/dist/x86_64-unknown-linux-gnu/rustup-init", false}, + {"/rustup/release-stable.toml", false}, + {"", false}, + // Near-misses must not be mistaken for a date segment. + {"/dist/20260801/x.tar.xz", false}, + {"/dist/v2026-08-01x/x.tar.xz", false}, + } + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + if got := isImmutableRustupPath(tt.path); got != tt.want { + t.Errorf("isImmutableRustupPath(%q) = %v, want %v", tt.path, got, tt.want) + } + }) + } +} + +func TestIsImmutableIndexPath(t *testing.T) { + // Every index path is mutable — a publish or a yank rewrites the entry. + for _, p := range []string{"/se/rd/serde", "/config.json", "/1/a"} { + if isImmutableIndexPath(p) { + t.Errorf("isImmutableIndexPath(%q) = true, want false", p) + } + } +} + +func TestExpandDL(t *testing.T) { + tests := []struct { + name string + tmpl string + crate string + version string + want string + wantOK bool + }{ + { + name: "crates.io marker template", + tmpl: "https://static.crates.io/crates/{crate}/{crate}-{version}.crate", + crate: "serde", + version: "1.0.203", + want: "https://static.crates.io/crates/serde/serde-1.0.203.crate", + wantOK: true, + }, + { + name: "marker-free template appends the default suffix", + tmpl: "https://static.crates.io/crates", + crate: "serde", + version: "1.0.203", + want: "https://static.crates.io/crates/serde/1.0.203/download", + wantOK: true, + }, + { + name: "marker-free template with trailing slash", + tmpl: "https://example.test/dl/", + crate: "tokio", + version: "1.2.3", + want: "https://example.test/dl/tokio/1.2.3/download", + wantOK: true, + }, + { + name: "prefix markers", + tmpl: "https://example.test/{prefix}/{crate}/{version}.crate", + crate: "serde", + version: "1.0.0", + want: "https://example.test/se/rd/serde/1.0.0.crate", + wantOK: true, + }, + { + name: "lowerprefix marker lowercases", + tmpl: "https://example.test/{lowerprefix}/{crate}", + crate: "Inflector", + version: "0.1.0", + want: "https://example.test/in/fl/Inflector", + wantOK: true, + }, + { + name: "short name prefix", + tmpl: "https://example.test/{prefix}/{crate}-{version}.crate", + crate: "cc", + version: "1.0.0", + want: "https://example.test/2/cc-1.0.0.crate", + wantOK: true, + }, + { + name: "sha256 marker is unusable", + tmpl: "https://example.test/{sha256-checksum}/{crate}", + crate: "serde", + version: "1.0.0", + wantOK: false, + }, + { + name: "empty template", + tmpl: "", + crate: "serde", + wantOK: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := expandDL(tt.tmpl, tt.crate, tt.version) + if ok != tt.wantOK { + t.Fatalf("expandDL ok = %v, want %v (got %q)", ok, tt.wantOK, got) + } + if ok && got != tt.want { + t.Errorf("expandDL = %q, want %q", got, tt.want) + } + }) + } +} + +func TestParseDLTemplate(t *testing.T) { + tests := []struct { + name string + raw string + want string + }{ + { + name: "real crates.io config", + raw: `{"dl":"https://static.crates.io/crates/{crate}/{crate}-{version}.crate","api":"https://crates.io"}`, + want: "https://static.crates.io/crates/{crate}/{crate}-{version}.crate", + }, + { + name: "marker-free dl", + raw: `{"dl":"https://example.test/crates","api":"https://example.test"}`, + want: "https://example.test/crates", + }, + {name: "malformed json falls back", raw: `{not json`, want: defaultDLTemplate}, + {name: "missing dl falls back", raw: `{"api":"https://crates.io"}`, want: defaultDLTemplate}, + {name: "blank dl falls back", raw: `{"dl":" "}`, want: defaultDLTemplate}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := parseDLTemplate([]byte(tt.raw)); got != tt.want { + t.Errorf("parseDLTemplate(%s) = %q, want %q", tt.raw, got, tt.want) + } + }) + } +} + +func TestRewriteConfigJSON(t *testing.T) { + raw := []byte(`{"dl":"https://static.crates.io/crates/{crate}/{crate}-{version}.crate","api":"https://crates.io","auth-required":false}`) + + out, err := rewriteConfigJSON(raw, "http://10.88.0.1:8083") + if err != nil { + t.Fatalf("rewriteConfigJSON: %v", err) + } + + var got map[string]any + if err := json.Unmarshal(out, &got); err != nil { + t.Fatalf("rewritten config is not valid JSON: %v", err) + } + + if got["dl"] != "http://10.88.0.1:8083/crates" { + t.Errorf("dl = %v, want the proxy's /crates route", got["dl"]) + } + // "api" must survive untouched so `cargo publish` and `cargo search` + // still reach the real registry. + if got["api"] != "https://crates.io" { + t.Errorf("api = %v, want https://crates.io (must not be rewritten)", got["api"]) + } + // Unknown keys must be preserved rather than dropped. + if _, ok := got["auth-required"]; !ok { + t.Error("rewrite dropped the unknown auth-required key") + } + + if _, err := rewriteConfigJSON([]byte("not json"), "http://x"); err == nil { + t.Error("rewriteConfigJSON accepted invalid JSON") + } +} + +// TestRewrittenDLRoundTrips pins the contract between the two halves of the +// crate route: whatever we advertise as "dl" must expand to a URL our own +// parseCrateDownloadPath can decode. +func TestRewrittenDLRoundTrips(t *testing.T) { + base := "http://10.88.0.1:8083" + out, err := rewriteConfigJSON([]byte(`{"dl":"https://static.crates.io/crates"}`), base) + if err != nil { + t.Fatalf("rewriteConfigJSON: %v", err) + } + tmpl := parseDLTemplate(out) + + url, ok := expandDL(tmpl, "serde", "1.0.203") + if !ok { + t.Fatal("expandDL rejected our own advertised template") + } + reqPath := strings.TrimPrefix(url, base) + + name, version, ok := parseCrateDownloadPath(reqPath) + if !ok { + t.Fatalf("parseCrateDownloadPath(%q) failed for our own route", reqPath) + } + if name != "serde" || version != "1.0.203" { + t.Errorf("round trip = (%q, %q), want (serde, 1.0.203)", name, version) + } +} + +func TestParseCrateDownloadPath(t *testing.T) { + tests := []struct { + path string + name, vers string + ok bool + }{ + {path: "/crates/serde/1.0.203/download", name: "serde", vers: "1.0.203", ok: true}, + {path: "/crates/proc-macro2/1.0.86/download", name: "proc-macro2", vers: "1.0.86", ok: true}, + {path: "/crates/tokio/1.0.0-beta.1/download", name: "tokio", vers: "1.0.0-beta.1", ok: true}, + + {path: "/crates/serde/1.0.203", ok: false}, + {path: "/crates/serde/1.0.203/fetch", ok: false}, + {path: "/crates/serde/1.0.203/download/extra", ok: false}, + {path: "/crates/serde", ok: false}, + {path: "/crates/", ok: false}, + {path: "/index/se/rd/serde", ok: false}, + {path: "/crates/../../etc/1.0.0/download", ok: false}, + {path: "/crates/se rde/1.0.0/download", ok: false}, + } + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + name, vers, ok := parseCrateDownloadPath(tt.path) + if ok != tt.ok { + t.Fatalf("parseCrateDownloadPath(%q) ok = %v, want %v", tt.path, ok, tt.ok) + } + if ok && (name != tt.name || vers != tt.vers) { + t.Errorf("= (%q, %q), want (%q, %q)", name, vers, tt.name, tt.vers) + } + }) + } +} + +func TestContainerConfigTOML(t *testing.T) { + got := containerConfigTOML("http://10.88.0.1:8083/") + + // Source replacement is the whole point — without both halves Cargo + // silently keeps using crates.io. + for _, want := range []string{ + "[source.crates-io]", + `replace-with = "ephemerd"`, + "[source.ephemerd]", + `registry = "sparse+http://10.88.0.1:8083/index/"`, + } { + if !strings.Contains(got, want) { + t.Errorf("generated config missing %q:\n%s", want, got) + } + } + // A doubled slash would produce an index URL Cargo cannot use. + if strings.Contains(got, "8083//index") { + t.Errorf("trailing slash in base leaked into the index URL:\n%s", got) + } +} + +func TestContainerConfigDest(t *testing.T) { + if got := containerConfigDest("linux"); got != "/.cargo" { + t.Errorf("containerConfigDest(linux) = %q, want /.cargo", got) + } + if got := containerConfigDest("darwin"); got != "/.cargo" { + t.Errorf("containerConfigDest(darwin) = %q, want /.cargo", got) + } + if got := containerConfigDest("windows"); got != `C:\.cargo` { + t.Errorf(`containerConfigDest(windows) = %q, want C:\.cargo`, got) + } +} + +func TestMatchesETag(t *testing.T) { + tests := []struct { + ifNoneMatch, tag string + want bool + }{ + {`"abc"`, `"abc"`, true}, + {`W/"abc"`, `"abc"`, true}, + {`"abc"`, `W/"abc"`, true}, + {`"x", "abc"`, `"abc"`, true}, + {`*`, `"abc"`, true}, + {`"abc"`, `"def"`, false}, + {``, `"abc"`, false}, + {`"abc"`, ``, false}, + {``, ``, false}, + } + for _, tt := range tests { + t.Run(tt.ifNoneMatch+"|"+tt.tag, func(t *testing.T) { + if got := matchesETag(tt.ifNoneMatch, tt.tag); got != tt.want { + t.Errorf("matchesETag(%q, %q) = %v, want %v", tt.ifNoneMatch, tt.tag, got, tt.want) + } + }) + } +} diff --git a/pkg/proxies/cargo/cargoproxy.go b/pkg/proxies/cargo/cargoproxy.go new file mode 100644 index 00000000..959c2d48 --- /dev/null +++ b/pkg/proxies/cargo/cargoproxy.go @@ -0,0 +1,752 @@ +// Package cargoproxy implements proxies.CacheProxy for the Rust ecosystem. +// +// It is a pull-through cache for three upstreams that a Rust CI job hits on +// every run: +// +// /index/… the crates.io SPARSE registry index (mutable) +// /crates/{c}/{v}/… .crate tarballs (immutable, content-addressed) +// /rustup/… rustup toolchain artifacts (dated ones immutable) +// +// ephemerd runs one shared instance on the bridge gateway IP so every job +// container can reach it. Jobs only ever issue HTTP GETs; they have no write +// access to the cache. +// +// Why both a mount and env vars: rustup takes its mirror from the +// environment (RUSTUP_DIST_SERVER), but Cargo does NOT — source replacement +// is only honoured from a config file, so the proxy generates a +// .cargo/config.toml on the host and declares it as a read-only mount at the +// container's filesystem root, where Cargo's ancestor-directory config search +// always finds it. See containerConfigTOML/containerConfigDest. +package cargoproxy + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "os" + "path/filepath" + goruntime "runtime" + "strings" + "sync" + "time" + + "github.com/ephpm/ephemerd/pkg/proxies" +) + +// hostGOOS is the OS this daemon runs on. Job containers share the host's +// OS (Linux jobs on Windows/macOS hosts run inside a Linux VM that has its +// own ephemerd), so it doubles as the default container OS. +const hostGOOS = goruntime.GOOS + +// Route prefixes. These are part of the contract with the generated +// container config (containerConfigTOML) and with the rewritten config.json. +const ( + indexRoute = "/index" + cratesRoute = "/crates" + rustupRoute = "/rustup" + healthRoute = "/healthz" +) + +// Defaults. Kept here rather than in the config package so the proxy is +// usable standalone (and testable) without a config file. +const ( + // DefaultIndexUpstream is the crates.io sparse index. + DefaultIndexUpstream = "https://index.crates.io" + // DefaultRustupUpstream is the rustup/toolchain distribution server. + DefaultRustupUpstream = "https://static.rust-lang.org" + // DefaultIndexTTL is how long a cached sparse-index entry is served + // without contacting upstream. Short by design: the index is mutable + // and a new dependency version must not take an hour to become + // visible. Revalidation after the TTL is a conditional GET, so the + // steady-state cost is a 304, not a re-download. + DefaultIndexTTL = 10 * time.Minute + // defaultTimeout bounds a single upstream request. + defaultTimeout = 120 * time.Second +) + +// Config for the Cargo caching proxy. +type Config struct { + // CacheDir is the on-disk cache root (e.g. /cache/cargo). + CacheDir string + // ConfDir is where the container-side Cargo config is generated. It is + // deliberately OUTSIDE CacheDir so `ephemerd cache clear cargo` cannot + // pull the mounted config out from under a running job. + ConfDir string + // IndexUpstream is the sparse registry index base URL. + IndexUpstream string + // RustupUpstream is the rustup distribution server base URL. + RustupUpstream string + // ListenAddr is the address to bind (e.g. "10.88.0.1:8083"). It is also + // the address advertised to containers — see advertiseBase. + ListenAddr string + // IndexTTL is the revalidation interval for mutable index entries. + // Zero means "unset" and takes DefaultIndexTTL; a NEGATIVE value means + // "revalidate on every request" (useful for tests and for operators who + // want the index never served without a conditional GET). + IndexTTL time.Duration + // Cleanup wipes the cache dir on Stop. Defaults to false: a + // pull-through cache that is emptied on every restart saves nothing. + Cleanup bool + // ContainerOS is the OS of the job containers this proxy serves, used + // only to pick the mount destination. Defaults to the host GOOS. + ContainerOS string + Log *slog.Logger +} + +// Compile-time interface checks. +var ( + _ proxies.CacheProxy = (*Proxy)(nil) + _ proxies.MountProvider = (*Proxy)(nil) +) + +// Proxy is a caching proxy for the Cargo registry and rustup distribution. +type Proxy struct { + cfg Config + server *http.Server + listener net.Listener + client *http.Client + inflight sync.Map // per-path mutex: collapses duplicate upstream fetches + + mu sync.RWMutex + // dlTemplate is the upstream registry's "dl" template, learned from + // config.json. Guarded because index requests (writers) and crate + // downloads (readers) run concurrently. + dlTemplate string +} + +// New creates a Cargo caching proxy. Call Start() to begin serving. +func New(cfg Config) *Proxy { + if cfg.IndexUpstream == "" { + cfg.IndexUpstream = DefaultIndexUpstream + } + if cfg.RustupUpstream == "" { + cfg.RustupUpstream = DefaultRustupUpstream + } + if cfg.IndexTTL == 0 { + cfg.IndexTTL = DefaultIndexTTL + } + if cfg.Log == nil { + cfg.Log = slog.Default() + } + cfg.IndexUpstream = strings.TrimRight(cfg.IndexUpstream, "/") + cfg.RustupUpstream = strings.TrimRight(cfg.RustupUpstream, "/") + + return &Proxy{ + cfg: cfg, + client: &http.Client{Timeout: defaultTimeout}, + dlTemplate: defaultDLTemplate, + } +} + +// Start begins serving the proxy. Returns after the listener is bound and +// the container-side Cargo config has been written. +func (p *Proxy) Start() error { + if err := os.MkdirAll(p.cfg.CacheDir, 0o755); err != nil { + return fmt.Errorf("creating cargo cache dir: %w", err) + } + if err := p.writeContainerConfig(); err != nil { + return fmt.Errorf("writing container cargo config: %w", err) + } + // Recover the upstream "dl" template from a previous run so the first + // crate download after a restart does not have to guess. + p.loadDLTemplateFromCache() + + // proxies.Listen, not net.Listen: the bridge gateway IP is not assigned + // to any interface until CNI creates the bridge for the first job. + ln, err := proxies.Listen(p.cfg.ListenAddr, p.cfg.Log) + if err != nil { + return err + } + p.listener = ln + + mux := http.NewServeMux() + mux.HandleFunc("/", p.handle) + p.server = &http.Server{Handler: mux, ReadHeaderTimeout: 30 * time.Second} + + go func() { + if err := p.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + p.cfg.Log.Error("cargo proxy server error", "error", err) + } + }() + + p.cfg.Log.Info("cargo proxy started", + "addr", ln.Addr().String(), + "advertised", p.advertiseBase(), + "cache", p.cfg.CacheDir, + "index_upstream", p.cfg.IndexUpstream, + "rustup_upstream", p.cfg.RustupUpstream, + "index_ttl", p.cfg.IndexTTL) + return nil +} + +// Stop shuts down the proxy and optionally wipes the cache. +func (p *Proxy) Stop() error { + var errs []error + + if p.server != nil { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := p.server.Shutdown(ctx); err != nil { + errs = append(errs, fmt.Errorf("shutting down cargo proxy: %w", err)) + } + } + + if p.cfg.Cleanup { + p.cfg.Log.Info("cleaning up cargo cache", "dir", p.cfg.CacheDir) + if err := os.RemoveAll(p.cfg.CacheDir); err != nil { + errs = append(errs, fmt.Errorf("cleaning cargo cache: %w", err)) + } + } + + return errors.Join(errs...) +} + +// Addr returns the address the proxy is listening on. +func (p *Proxy) Addr() string { + if p.listener != nil { + return p.listener.Addr().String() + } + return p.cfg.ListenAddr +} + +// advertiseBase is the base URL containers use to reach this proxy. It is +// built from the CONFIGURED address, not the bound one: after a wildcard +// fallback the bound address is "[::]:8083", which means nothing inside a +// container. +func (p *Proxy) advertiseBase() string { + addr := p.cfg.ListenAddr + if addr == "" { + addr = p.Addr() + } + return "http://" + addr +} + +// EnvVars returns environment variables to inject into job containers. +// +// Only rustup is configured here. Cargo's registry redirect cannot be done +// with environment variables (see Mounts). +func (p *Proxy) EnvVars() []string { + return []string{ + "RUSTUP_DIST_SERVER=" + p.advertiseBase() + rustupRoute, + } +} + +// Mounts returns the read-only bind mount that carries the generated Cargo +// config into job containers. +func (p *Proxy) Mounts() []proxies.Mount { + return []proxies.Mount{{ + Source: p.hostConfigDir(), + Destination: containerConfigDest(p.containerOS()), + ReadOnly: true, + }} +} + +// Name returns the proxy name for logging. +func (p *Proxy) Name() string { return "cargo" } + +func (p *Proxy) containerOS() string { + if p.cfg.ContainerOS != "" { + return p.cfg.ContainerOS + } + return hostGOOS +} + +// hostConfigDir is the host-side directory that is mounted as the +// container's ".cargo" directory. +func (p *Proxy) hostConfigDir() string { + return filepath.Join(p.cfg.ConfDir, ".cargo") +} + +// writeContainerConfig regenerates the container-side Cargo config. Written +// atomically so a job that starts mid-write never sees a truncated file. +func (p *Proxy) writeContainerConfig() error { + dir := p.hostConfigDir() + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("creating cargo conf dir %q: %w", dir, err) + } + target := filepath.Join(dir, "config.toml") + tmp, err := os.CreateTemp(dir, ".config-*.toml") + if err != nil { + return fmt.Errorf("creating temp cargo config: %w", err) + } + if _, err := tmp.WriteString(containerConfigTOML(p.advertiseBase())); err != nil { + _ = tmp.Close() + _ = os.Remove(tmp.Name()) + return fmt.Errorf("writing temp cargo config: %w", err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("closing temp cargo config: %w", err) + } + if err := os.Chmod(tmp.Name(), 0o644); err != nil { + p.cfg.Log.Debug("chmod cargo config", "error", err) + } + if err := os.Rename(tmp.Name(), target); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("installing cargo config at %q: %w", target, err) + } + p.cfg.Log.Info("wrote container cargo config", "path", target, "mount", containerConfigDest(p.containerOS())) + return nil +} + +// --- routing --------------------------------------------------------------- + +func (p *Proxy) handle(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + path := r.URL.Path + switch { + case path == healthRoute: + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "ok\n") + + case path == indexRoute+"/config.json": + p.handleRegistryConfig(w, r) + + case strings.HasPrefix(path, indexRoute+"/"): + p.handleIndex(w, r, strings.TrimPrefix(path, indexRoute)) + + case strings.HasPrefix(path, cratesRoute+"/"): + p.handleCrate(w, r) + + case strings.HasPrefix(path, rustupRoute+"/"): + p.handleRustup(w, r, strings.TrimPrefix(path, rustupRoute)) + + default: + http.NotFound(w, r) + } +} + +// handleRegistryConfig serves the registry's config.json with "dl" rewritten +// to point crate downloads at this proxy. The upstream copy is cached (and +// revalidated) like any other index entry, and its "dl" template is retained +// so crate requests can be mapped back to the real CDN. +func (p *Proxy) handleRegistryConfig(w http.ResponseWriter, r *http.Request) { + cachePath, err := indexCachePath(p.cfg.CacheDir, "/config.json") + if err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + body, _, err := p.fetchCached(r.Context(), cachePath, p.cfg.IndexUpstream+"/config.json", false, p.cfg.IndexTTL) + if err != nil { + p.cfg.Log.Warn("registry config.json unavailable", "error", err) + http.Error(w, "upstream error", http.StatusBadGateway) + return + } + + p.setDLTemplate(parseDLTemplate(body)) + + rewritten, err := rewriteConfigJSON(body, p.advertiseBase()) + if err != nil { + // Cannot rewrite: send the upstream config unchanged so cargo + // downloads straight from the CDN rather than failing. + p.cfg.Log.Warn("serving upstream registry config unrewritten", "error", err) + rewritten = body + } + + w.Header().Set("Content-Type", "application/json") + // No ETag: the body is ours, not upstream's, so upstream validators + // would be wrong. The document is tiny. + w.Header().Set("Cache-Control", "no-cache") + writeBody(w, r, rewritten) +} + +// handleIndex serves a sparse-index entry. Index data is MUTABLE, so a +// cached entry is served for IndexTTL and then revalidated with a +// conditional GET. +func (p *Proxy) handleIndex(w http.ResponseWriter, r *http.Request, indexPath string) { + cachePath, err := indexCachePath(p.cfg.CacheDir, indexPath) + if err != nil { + p.cfg.Log.Warn("rejecting index path", "path", indexPath, "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + body, meta, err := p.fetchCached(r.Context(), cachePath, p.cfg.IndexUpstream+indexPath, isImmutableIndexPath(indexPath), p.cfg.IndexTTL) + if err != nil { + if isNotFound(err) { + http.NotFound(w, r) + return + } + // Nothing cached and upstream is unreachable. Cargo treats this as + // fatal either way; report it honestly. + p.cfg.Log.Warn("index fetch failed", "path", indexPath, "error", err) + http.Error(w, "upstream error", http.StatusBadGateway) + return + } + + // Cheap client-side revalidation: Cargo keeps its own copy in CARGO_HOME + // and sends If-None-Match, so most warm-cache requests can be a 304. + if meta.ETag != "" { + w.Header().Set("ETag", meta.ETag) + if matchesETag(r.Header.Get("If-None-Match"), meta.ETag) { + w.WriteHeader(http.StatusNotModified) + return + } + } + if meta.LastModified != "" { + w.Header().Set("Last-Modified", meta.LastModified) + } + w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, "text/plain; charset=utf-8")) + writeBody(w, r, body) +} + +// handleCrate serves a .crate tarball. Tarballs are IMMUTABLE — a published +// (crate, version) is never republished with different bytes — so a cached +// file is served forever and never revalidated. +func (p *Proxy) handleCrate(w http.ResponseWriter, r *http.Request) { + name, version, ok := parseCrateDownloadPath(r.URL.Path) + if !ok { + p.cfg.Log.Warn("rejecting crate download path", "path", r.URL.Path) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + cachePath, err := crateCachePath(p.cfg.CacheDir, name, version) + if err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + upstreamURL, ok := expandDL(p.dlTemplateOrDefault(), name, version) + if !ok { + upstreamURL, _ = expandDL(defaultDLTemplate, name, version) + } + + body, _, err := p.fetchCached(r.Context(), cachePath, upstreamURL, true, 0) + if err != nil { + if isNotFound(err) { + http.NotFound(w, r) + return + } + // FAIL OPEN: we could not fetch it, but the container can reach the + // internet directly. Redirect to the real CDN so the build proceeds + // (slower, uncached) instead of failing. + p.cfg.Log.Warn("crate fetch failed; redirecting job to upstream", + "crate", name, "version", version, "upstream", upstreamURL, "error", err) + http.Redirect(w, r, upstreamURL, http.StatusTemporaryRedirect) + return + } + + w.Header().Set("Content-Type", "application/x-tar") + // Immutable by construction — let Cargo and any intermediary cache hard. + w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") + writeBody(w, r, body) +} + +// handleRustup serves rustup distribution artifacts. Dated artifacts are +// immutable; channel manifests are revalidated on the index TTL. +func (p *Proxy) handleRustup(w http.ResponseWriter, r *http.Request, distPath string) { + cachePath, err := rustupCachePath(p.cfg.CacheDir, distPath) + if err != nil { + p.cfg.Log.Warn("rejecting rustup path", "path", distPath, "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + upstreamURL := p.cfg.RustupUpstream + distPath + immutable := isImmutableRustupPath(distPath) + + body, meta, err := p.fetchCached(r.Context(), cachePath, upstreamURL, immutable, p.cfg.IndexTTL) + if err != nil { + if isNotFound(err) { + http.NotFound(w, r) + return + } + // FAIL OPEN, same reasoning as crate tarballs. + p.cfg.Log.Warn("rustup fetch failed; redirecting job to upstream", + "path", distPath, "error", err) + http.Redirect(w, r, upstreamURL, http.StatusTemporaryRedirect) + return + } + + if meta.ETag != "" { + w.Header().Set("ETag", meta.ETag) + } + w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, "application/octet-stream")) + if immutable { + w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") + } + writeBody(w, r, body) +} + +// --- cache core ------------------------------------------------------------ + +// notFoundError marks an upstream 404/410 so handlers can pass it through as +// a 404 rather than treating it as an outage. +type notFoundError struct{ url string } + +func (e *notFoundError) Error() string { return "upstream 404 for " + e.url } + +func isNotFound(err error) bool { + var nf *notFoundError + return errors.As(err, &nf) +} + +// fetchCached is the single read path for every route. +// +// It resolves the freshness decision (decideFreshness), then: +// +// - serveCached → return the on-disk bytes, no network at all; +// - revalidate → conditional GET; 304 refreshes the timestamp and the +// cached bytes are returned; 200 replaces them; +// - fetchFresh → unconditional GET, then store. +// +// FAIL-OPEN AT THIS LAYER: if the upstream request errors or returns 5xx and +// we HAVE a cached copy, the stale copy is served rather than propagating +// the failure. A registry outage then degrades to "slightly stale index" +// instead of a red CI job. Callers layer a second fail-open on top (a +// redirect to the origin) for the case where nothing is cached. +func (p *Proxy) fetchCached(ctx context.Context, cachePath, upstreamURL string, immutable bool, ttl time.Duration) ([]byte, entryMeta, error) { + // Fast path: a fresh (or immutable) hit needs no lock and no network. + if body, meta, ok := p.readCache(cachePath); ok { + if decideFreshness(true, immutable, meta.Fetched, time.Now(), ttl) == serveCached { + p.cfg.Log.Debug("cargo cache hit", "url", upstreamURL, "immutable", immutable) + return body, meta, nil + } + } + + // Collapse concurrent misses for the same object: without this, ten + // parallel jobs starting together each pull the same 40 MB toolchain. + mu := p.lockFor(cachePath) + mu.Lock() + defer mu.Unlock() + + body, meta, cached := p.readCache(cachePath) + decision := decideFreshness(cached, immutable, meta.Fetched, time.Now(), ttl) + if decision == serveCached { + p.cfg.Log.Debug("cargo cache hit (after lock)", "url", upstreamURL) + return body, meta, nil + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) + if err != nil { + if cached { + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("building upstream request for %s: %w", upstreamURL, err) + } + if decision == revalidate { + if meta.ETag != "" { + req.Header.Set("If-None-Match", meta.ETag) + } + if meta.LastModified != "" { + req.Header.Set("If-Modified-Since", meta.LastModified) + } + } + + p.cfg.Log.Debug("cargo cache miss", "url", upstreamURL, "decision", decision.String()) + resp, err := p.client.Do(req) + if err != nil { + if cached { + p.cfg.Log.Warn("upstream unreachable; serving stale cache", "url", upstreamURL, "error", err) + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("fetching %s: %w", upstreamURL, err) + } + defer func() { + if cerr := resp.Body.Close(); cerr != nil { + p.cfg.Log.Debug("closing upstream body", "error", cerr) + } + }() + + switch { + case resp.StatusCode == http.StatusNotModified && cached: + meta.Fetched = time.Now() + p.writeMeta(cachePath, meta) + return body, meta, nil + + case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone: + return nil, entryMeta{}, ¬FoundError{url: upstreamURL} + + case resp.StatusCode != http.StatusOK: + if cached { + p.cfg.Log.Warn("upstream error; serving stale cache", + "url", upstreamURL, "status", resp.StatusCode) + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("fetching %s: upstream status %d", upstreamURL, resp.StatusCode) + } + + fresh, err := io.ReadAll(resp.Body) + if err != nil { + if cached { + p.cfg.Log.Warn("upstream read failed; serving stale cache", "url", upstreamURL, "error", err) + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("reading %s: %w", upstreamURL, err) + } + + newMeta := entryMeta{ + ETag: resp.Header.Get("ETag"), + LastModified: resp.Header.Get("Last-Modified"), + ContentType: resp.Header.Get("Content-Type"), + Fetched: time.Now(), + } + // A write failure must not fail the request — the bytes are already in + // hand, the job just loses the caching benefit. + if err := p.writeCache(cachePath, fresh, newMeta, immutable); err != nil { + p.cfg.Log.Warn("caching response failed; serving anyway", "url", upstreamURL, "error", err) + } + return fresh, newMeta, nil +} + +// readCache loads a cached body and its sidecar metadata. A body with no +// sidecar is still usable (it reports a zero Fetched time, so mutable +// entries revalidate and immutable ones are served as-is). +func (p *Proxy) readCache(cachePath string) ([]byte, entryMeta, bool) { + body, err := os.ReadFile(cachePath) + if err != nil { + return nil, entryMeta{}, false + } + var meta entryMeta + if raw, err := os.ReadFile(metaPath(cachePath)); err == nil { + if err := json.Unmarshal(raw, &meta); err != nil { + p.cfg.Log.Debug("discarding unreadable cache metadata", "path", cachePath, "error", err) + meta = entryMeta{} + } + } + return body, meta, true +} + +// writeCache stores a body and its metadata atomically (write temp, rename) +// so a concurrent reader never sees a partial file. Immutable entries get no +// sidecar: there is nothing to revalidate, and skipping it halves the inode +// count for the tarball cache. +func (p *Proxy) writeCache(cachePath string, body []byte, meta entryMeta, immutable bool) error { + dir := filepath.Dir(cachePath) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("creating cache dir %q: %w", dir, err) + } + if err := atomicWrite(cachePath, body); err != nil { + return err + } + if immutable { + return nil + } + p.writeMeta(cachePath, meta) + return nil +} + +// writeMeta persists the sidecar. Best-effort: a missing sidecar only costs +// a revalidation next time. +func (p *Proxy) writeMeta(cachePath string, meta entryMeta) { + raw, err := json.Marshal(meta) + if err != nil { + p.cfg.Log.Debug("encoding cache metadata", "path", cachePath, "error", err) + return + } + if err := atomicWrite(metaPath(cachePath), raw); err != nil { + p.cfg.Log.Debug("writing cache metadata", "path", cachePath, "error", err) + } +} + +func atomicWrite(target string, data []byte) error { + dir := filepath.Dir(target) + tmp, err := os.CreateTemp(dir, ".cargoproxy-*") + if err != nil { + return fmt.Errorf("creating temp file in %q: %w", dir, err) + } + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + _ = os.Remove(tmp.Name()) + return fmt.Errorf("writing %q: %w", target, err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("closing temp file for %q: %w", target, err) + } + if err := os.Rename(tmp.Name(), target); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("renaming temp file to %q: %w", target, err) + } + return nil +} + +func (p *Proxy) lockFor(key string) *sync.Mutex { + v, _ := p.inflight.LoadOrStore(key, &sync.Mutex{}) + return v.(*sync.Mutex) +} + +func (p *Proxy) setDLTemplate(tmpl string) { + if tmpl == "" { + return + } + p.mu.Lock() + p.dlTemplate = tmpl + p.mu.Unlock() +} + +func (p *Proxy) dlTemplateOrDefault() string { + p.mu.RLock() + defer p.mu.RUnlock() + if p.dlTemplate == "" { + return defaultDLTemplate + } + return p.dlTemplate +} + +// loadDLTemplateFromCache recovers the upstream "dl" template across a +// daemon restart. Cargo caches config.json in CARGO_HOME too, so it may go +// straight to a crate download without asking us for config.json first. +func (p *Proxy) loadDLTemplateFromCache() { + cachePath, err := indexCachePath(p.cfg.CacheDir, "/config.json") + if err != nil { + return + } + raw, err := os.ReadFile(cachePath) + if err != nil { + return + } + p.setDLTemplate(parseDLTemplate(raw)) +} + +// --- small helpers --------------------------------------------------------- + +// writeBody writes the response body, honouring HEAD (headers only). +func writeBody(w http.ResponseWriter, r *http.Request, body []byte) { + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body))) + if r.Method == http.MethodHead { + w.WriteHeader(http.StatusOK) + return + } + w.WriteHeader(http.StatusOK) + if _, err := w.Write(body); err != nil { + // The client went away mid-response; nothing to do but note it. + _ = err + } +} + +// matchesETag reports whether an If-None-Match header matches the given tag. +// Handles the "*" wildcard, comma-separated lists, and the W/ weak prefix. +func matchesETag(ifNoneMatch, tag string) bool { + if ifNoneMatch == "" || tag == "" { + return false + } + if strings.TrimSpace(ifNoneMatch) == "*" { + return true + } + want := strings.TrimPrefix(tag, "W/") + for _, candidate := range strings.Split(ifNoneMatch, ",") { + if strings.TrimPrefix(strings.TrimSpace(candidate), "W/") == want { + return true + } + } + return false +} + +func contentTypeOr(got, fallback string) string { + if got == "" { + return fallback + } + return got +} diff --git a/pkg/proxies/cargo/cargoproxy_test.go b/pkg/proxies/cargo/cargoproxy_test.go new file mode 100644 index 00000000..10a18e17 --- /dev/null +++ b/pkg/proxies/cargo/cargoproxy_test.go @@ -0,0 +1,678 @@ +package cargoproxy + +import ( + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// NOTE: every test here runs against an httptest fake upstream. Nothing in +// this file touches crates.io, static.rust-lang.org, or any other network. + +// alwaysRevalidate is a negative TTL, which Config documents as "never serve +// a mutable entry without a conditional GET". Tests use it to reach the +// aged-out state deterministically instead of sleeping. A ZERO TTL will not +// do: New() reads zero as "unset" and substitutes DefaultIndexTTL. +const alwaysRevalidate = -1 * time.Second + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// fakeUpstream is a stand-in for index.crates.io / static.crates.io / +// static.rust-lang.org, with per-path hit counting so tests can assert that +// the cache actually prevented a second fetch. +type fakeUpstream struct { + *httptest.Server + + mu sync.Mutex + hits map[string]int + bodies map[string]string + etags map[string]string + status map[string]int + failing atomic.Bool + condHits atomic.Int64 // conditional (If-None-Match) requests received +} + +func newFakeUpstream(t *testing.T) *fakeUpstream { + t.Helper() + f := &fakeUpstream{ + hits: map[string]int{}, + bodies: map[string]string{}, + etags: map[string]string{}, + status: map[string]int{}, + } + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.hits[r.URL.Path]++ + body, known := f.bodies[r.URL.Path] + etag := f.etags[r.URL.Path] + status := f.status[r.URL.Path] + f.mu.Unlock() + + if f.failing.Load() { + http.Error(w, "upstream down", http.StatusInternalServerError) + return + } + if status != 0 { + w.WriteHeader(status) + return + } + if !known { + http.NotFound(w, r) + return + } + if inm := r.Header.Get("If-None-Match"); inm != "" { + f.condHits.Add(1) + if etag != "" && inm == etag { + w.Header().Set("ETag", etag) + w.WriteHeader(http.StatusNotModified) + return + } + } + if etag != "" { + w.Header().Set("ETag", etag) + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = io.WriteString(w, body) + })) + t.Cleanup(f.Close) + return f +} + +func (f *fakeUpstream) set(path, body, etag string) { + f.mu.Lock() + defer f.mu.Unlock() + f.bodies[path] = body + if etag != "" { + f.etags[path] = etag + } +} + +func (f *fakeUpstream) setStatus(path string, code int) { + f.mu.Lock() + defer f.mu.Unlock() + f.status[path] = code +} + +func (f *fakeUpstream) hitCount(path string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.hits[path] +} + +// newTestProxy starts a proxy bound to loopback with both upstreams pointed +// at the fake, and returns it plus its base URL. +func newTestProxy(t *testing.T, up *fakeUpstream, mutate func(*Config)) (*Proxy, string) { + t.Helper() + dir := t.TempDir() + cfg := Config{ + CacheDir: filepath.Join(dir, "cache"), + ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, + RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", + IndexTTL: time.Hour, + ContainerOS: "linux", + Log: discardLogger(), + } + if mutate != nil { + mutate(&cfg) + } + p := New(cfg) + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { + if err := p.Stop(); err != nil { + t.Errorf("Stop: %v", err) + } + }) + return p, "http://" + p.Addr() +} + +func get(t *testing.T, url string, hdr map[string]string) (*http.Response, string) { + t.Helper() + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + t.Fatalf("building request for %s: %v", url, err) + } + for k, v := range hdr { + req.Header.Set(k, v) + } + // Do not follow redirects: the fail-open path is a 307 we want to assert. + client := &http.Client{ + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + Timeout: 10 * time.Second, + } + resp, err := client.Do(req) + if err != nil { + t.Fatalf("GET %s: %v", url, err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("reading body of %s: %v", url, err) + } + return resp, string(body) +} + +// --- index (mutable) ------------------------------------------------------- + +// TestIndex_CachedWithinTTL: the sparse index is mutable, but inside the TTL +// it must be served from disk with no upstream request at all. +func TestIndex_CachedWithinTTL(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", `{"name":"serde","vers":"1.0.203"}`, `"v1"`) + _, base := newTestProxy(t, up, nil) + + for i := range 3 { + resp, body := get(t, base+"/index/se/rd/serde", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("request %d: status = %d, want 200", i, resp.StatusCode) + } + if !strings.Contains(body, `"serde"`) { + t.Fatalf("request %d: body = %q", i, body) + } + } + if got := up.hitCount("/se/rd/serde"); got != 1 { + t.Errorf("upstream hits = %d, want 1 (later requests must come from cache)", got) + } +} + +// TestIndex_RevalidatesAfterTTL: past the TTL the proxy must NOT serve stale +// data blindly — it must issue a conditional GET, and it must pick up new +// content when the entry actually changed. +func TestIndex_RevalidatesAfterTTL(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "v1-body", `"v1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "v1-body" { + t.Fatalf("first body = %q, want v1-body", body) + } + + // Unchanged upstream: the revalidation must be conditional and the + // cached bytes reused. + before := up.condHits.Load() + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "v1-body" { + t.Fatalf("second body = %q, want v1-body", body) + } + if up.condHits.Load() <= before { + t.Error("revalidation was not conditional (no If-None-Match sent)") + } + + // Changed upstream: a new version was published, the proxy must see it. + up.set("/se/rd/serde", "v2-body", `"v2"`) + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "v2-body" { + t.Errorf("after upstream change body = %q, want v2-body (stale index served)", body) + } +} + +// TestIndex_ServesClient304 keeps warm-cache traffic cheap: Cargo keeps its +// own copy and sends If-None-Match, which must produce a bodyless 304. +func TestIndex_ServesClient304(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "body", `"v1"`) + _, base := newTestProxy(t, up, nil) + + resp, _ := get(t, base+"/index/se/rd/serde", nil) + etag := resp.Header.Get("ETag") + if etag == "" { + t.Fatal("proxy did not surface the upstream ETag") + } + + resp2, body2 := get(t, base+"/index/se/rd/serde", map[string]string{"If-None-Match": etag}) + if resp2.StatusCode != http.StatusNotModified { + t.Errorf("status = %d, want 304", resp2.StatusCode) + } + if body2 != "" { + t.Errorf("304 carried a body: %q", body2) + } +} + +// TestIndex_FailOpenToStaleCache: a registry outage must not turn into a red +// CI job when we already hold the data. +func TestIndex_FailOpenToStaleCache(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "cached-body", `"v1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "cached-body" { + t.Fatalf("priming failed: %q", body) + } + + up.failing.Store(true) + resp, body := get(t, base+"/index/se/rd/serde", nil) + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want 200 (stale cache should be served)", resp.StatusCode) + } + if body != "cached-body" { + t.Errorf("body = %q, want the stale cached copy", body) + } +} + +// TestIndex_UpstreamNotFoundIsPassedThrough: a crate that does not exist is +// a 404, not a 502 — Cargo needs to distinguish them. +func TestIndex_UpstreamNotFoundIsPassedThrough(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + resp, _ := get(t, base+"/index/no/pe/nope", nil) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 404", resp.StatusCode) + } +} + +// TestIndex_RejectsTraversal keeps a hostile request from writing outside +// the cache root. +func TestIndex_RejectsTraversal(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + // http.Client normalises "..", so build the raw request by hand. + resp, _ := get(t, base+"/index/se/rd/%2e%2e%2f%2e%2e%2fescape", nil) + if resp.StatusCode != http.StatusBadRequest && resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 400 or 404 for a traversal attempt", resp.StatusCode) + } +} + +// --- config.json rewrite --------------------------------------------------- + +func TestRegistryConfig_RewritesDLToProxy(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+"UPSTREAM"+`/crates/{crate}/{crate}-{version}.crate","api":"https://crates.io"}`, "") + p, base := newTestProxy(t, up, nil) + + resp, body := get(t, base+"/index/config.json", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + if !strings.Contains(body, `"dl":"`+p.advertiseBase()+`/crates"`) { + t.Errorf("dl was not rewritten to the proxy: %s", body) + } + if !strings.Contains(body, `"api":"https://crates.io"`) { + t.Errorf("api must be preserved so cargo publish still works: %s", body) + } +} + +// --- crate tarballs (immutable) -------------------------------------------- + +func TestCrate_CachedForeverAfterFirstFetch(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + up.set("/crates/serde/serde-1.0.203.crate", "TARBALL-BYTES", `"e1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + // Cargo asks for config.json first; that is what teaches us the dl template. + if resp, _ := get(t, base+"/index/config.json", nil); resp.StatusCode != http.StatusOK { + t.Fatalf("config.json status = %d", resp.StatusCode) + } + + for i := range 3 { + resp, body := get(t, base+"/crates/serde/1.0.203/download", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("request %d: status = %d, want 200", i, resp.StatusCode) + } + if body != "TARBALL-BYTES" { + t.Fatalf("request %d: body = %q", i, body) + } + } + + // Immutable content: exactly one upstream fetch, and no revalidation + // even though the index TTL is zero. + if got := up.hitCount("/crates/serde/serde-1.0.203.crate"); got != 1 { + t.Errorf("upstream hits = %d, want 1 — .crate tarballs are immutable and must never be refetched", got) + } +} + +// TestCrate_ImmutableCacheSurvivesUpstreamOutage: once cached, a tarball is +// served with the network completely gone. +func TestCrate_ImmutableCacheSurvivesUpstreamOutage(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + up.set("/crates/serde/serde-1.0.203.crate", "TARBALL-BYTES", "") + _, base := newTestProxy(t, up, nil) + + get(t, base+"/index/config.json", nil) + get(t, base+"/crates/serde/1.0.203/download", nil) + + up.failing.Store(true) + resp, body := get(t, base+"/crates/serde/1.0.203/download", nil) + if resp.StatusCode != http.StatusOK || body != "TARBALL-BYTES" { + t.Errorf("status = %d body = %q, want a cache hit despite the outage", resp.StatusCode, body) + } +} + +// TestCrate_FailsOpenWithRedirect is the core fail-open guarantee: when the +// proxy cannot produce the tarball, the job must be sent to the real CDN +// rather than handed an error. +func TestCrate_FailsOpenWithRedirect(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + _, base := newTestProxy(t, up, nil) + get(t, base+"/index/config.json", nil) + + // 500 from upstream with nothing cached. + up.setStatus("/crates/serde/serde-9.9.9.crate", http.StatusInternalServerError) + + resp, _ := get(t, base+"/crates/serde/9.9.9/download", nil) + if resp.StatusCode != http.StatusTemporaryRedirect { + t.Fatalf("status = %d, want 307 (fail open to upstream)", resp.StatusCode) + } + loc := resp.Header.Get("Location") + if !strings.HasSuffix(loc, "/crates/serde/serde-9.9.9.crate") { + t.Errorf("Location = %q, want the upstream tarball URL", loc) + } +} + +// TestCrate_UnknownVersionIs404 keeps a genuine "no such version" distinct +// from an outage — redirecting on a 404 would just make Cargo fetch a 404. +func TestCrate_UnknownVersionIs404(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + _, base := newTestProxy(t, up, nil) + get(t, base+"/index/config.json", nil) + + resp, _ := get(t, base+"/crates/serde/0.0.0/download", nil) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 404", resp.StatusCode) + } +} + +func TestCrate_RejectsMalformedPaths(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + for _, p := range []string{ + "/crates/serde/1.0.0", + "/crates/serde/1.0.0/fetch", + "/crates/serde%20x/1.0.0/download", + } { + resp, _ := get(t, base+p, nil) + if resp.StatusCode == http.StatusOK { + t.Errorf("GET %s = 200, want a rejection", p) + } + } +} + +// TestCrate_ConcurrentMissesFetchUpstreamOnce pins the single-flight +// behaviour that stops N parallel jobs pulling the same tarball N times. +func TestCrate_ConcurrentMissesFetchUpstreamOnce(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + up.set("/crates/tokio/tokio-1.2.3.crate", "BIG-TARBALL", "") + _, base := newTestProxy(t, up, nil) + get(t, base+"/index/config.json", nil) + + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + resp, err := http.Get(base + "/crates/tokio/1.2.3/download") + if err != nil { + return + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + }() + } + wg.Wait() + + if got := up.hitCount("/crates/tokio/tokio-1.2.3.crate"); got != 1 { + t.Errorf("upstream hits = %d, want 1 (concurrent misses must be collapsed)", got) + } +} + +// --- rustup ---------------------------------------------------------------- + +func TestRustup_DatedArtifactCachedForever(t *testing.T) { + up := newFakeUpstream(t) + up.set("/dist/2026-08-01/rust-std-nightly.tar.xz", "TOOLCHAIN", `"e1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + for range 3 { + resp, body := get(t, base+"/rustup/dist/2026-08-01/rust-std-nightly.tar.xz", nil) + if resp.StatusCode != http.StatusOK || body != "TOOLCHAIN" { + t.Fatalf("status = %d body = %q", resp.StatusCode, body) + } + } + if got := up.hitCount("/dist/2026-08-01/rust-std-nightly.tar.xz"); got != 1 { + t.Errorf("upstream hits = %d, want 1 (dated toolchain artifacts are immutable)", got) + } +} + +func TestRustup_ChannelManifestRevalidates(t *testing.T) { + up := newFakeUpstream(t) + up.set("/dist/channel-rust-nightly.toml", "manifest-v1", `"m1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + if _, body := get(t, base+"/rustup/dist/channel-rust-nightly.toml", nil); body != "manifest-v1" { + t.Fatalf("first fetch = %q", body) + } + // The channel manifest is rewritten daily in place; the proxy must not + // pin jobs to yesterday's nightly. + up.set("/dist/channel-rust-nightly.toml", "manifest-v2", `"m2"`) + if _, body := get(t, base+"/rustup/dist/channel-rust-nightly.toml", nil); body != "manifest-v2" { + t.Errorf("second fetch = %q, want manifest-v2 (stale channel manifest served)", body) + } +} + +func TestRustup_FailsOpenWithRedirect(t *testing.T) { + up := newFakeUpstream(t) + up.setStatus("/dist/2026-08-01/x.tar.xz", http.StatusInternalServerError) + _, base := newTestProxy(t, up, nil) + + resp, _ := get(t, base+"/rustup/dist/2026-08-01/x.tar.xz", nil) + if resp.StatusCode != http.StatusTemporaryRedirect { + t.Fatalf("status = %d, want 307 (fail open)", resp.StatusCode) + } + if want := up.URL + "/dist/2026-08-01/x.tar.xz"; resp.Header.Get("Location") != want { + t.Errorf("Location = %q, want %q", resp.Header.Get("Location"), want) + } +} + +// --- lifecycle / wiring ---------------------------------------------------- + +// TestStart_WritesContainerConfigAndDeclaresMount is what makes the proxy +// actually get used: without the generated file (and the mount that carries +// it) Cargo silently keeps talking to crates.io. +func TestStart_WritesContainerConfigAndDeclaresMount(t *testing.T) { + up := newFakeUpstream(t) + p, _ := newTestProxy(t, up, nil) + + mounts := p.Mounts() + if len(mounts) != 1 { + t.Fatalf("Mounts() = %d entries, want 1", len(mounts)) + } + m := mounts[0] + if m.Destination != "/.cargo" { + t.Errorf("mount destination = %q, want /.cargo", m.Destination) + } + if !m.ReadOnly { + t.Error("the cargo config mount must be read-only — a job must not rewrite the next job's config") + } + + raw, err := os.ReadFile(filepath.Join(m.Source, "config.toml")) + if err != nil { + t.Fatalf("reading generated config: %v", err) + } + if !strings.Contains(string(raw), `replace-with = "ephemerd"`) { + t.Errorf("generated config is not a source replacement:\n%s", raw) + } + if !strings.Contains(string(raw), "sparse+"+p.advertiseBase()+"/index/") { + t.Errorf("generated config does not point at this proxy:\n%s", raw) + } +} + +// TestConfigDirIsOutsideCacheDir: `ephemerd cache clear cargo` wipes the +// cache root. If the mounted config lived there it would vanish from under a +// running job. +func TestConfigDirIsOutsideCacheDir(t *testing.T) { + up := newFakeUpstream(t) + p, _ := newTestProxy(t, up, nil) + + confAbs, err := filepath.Abs(p.Mounts()[0].Source) + if err != nil { + t.Fatalf("abs conf dir: %v", err) + } + cacheAbs, err := filepath.Abs(p.cfg.CacheDir) + if err != nil { + t.Fatalf("abs cache dir: %v", err) + } + rel, err := filepath.Rel(cacheAbs, confAbs) + if err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + t.Errorf("container config dir %q is inside the clearable cache dir %q", confAbs, cacheAbs) + } +} + +func TestEnvVars_PointRustupAtTheProxy(t *testing.T) { + up := newFakeUpstream(t) + p, _ := newTestProxy(t, up, nil) + + env := p.EnvVars() + if len(env) != 1 { + t.Fatalf("EnvVars() = %v, want exactly the rustup dist server", env) + } + want := "RUSTUP_DIST_SERVER=" + p.advertiseBase() + "/rustup" + if env[0] != want { + t.Errorf("EnvVars()[0] = %q, want %q", env[0], want) + } +} + +// TestAdvertiseBase_UsesConfiguredAddress: after a wildcard fallback the +// bound address is "[::]:port", which is useless inside a container. The +// advertised address must stay the configured gateway address. +func TestAdvertiseBase_UsesConfiguredAddress(t *testing.T) { + p := New(Config{ListenAddr: "10.88.0.1:8083", Log: discardLogger()}) + if got := p.advertiseBase(); got != "http://10.88.0.1:8083" { + t.Errorf("advertiseBase() = %q, want http://10.88.0.1:8083", got) + } +} + +func TestStop_CleanupHonoursConfig(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "body", "") + + t.Run("cleanup disabled keeps the cache", func(t *testing.T) { + dir := t.TempDir() + p := New(Config{ + CacheDir: filepath.Join(dir, "cache"), ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", ContainerOS: "linux", Cleanup: false, Log: discardLogger(), + }) + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + get(t, "http://"+p.Addr()+"/index/se/rd/serde", nil) + if err := p.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "cache", "index", "se", "rd", "serde")); err != nil { + t.Errorf("cache was wiped despite cleanup=false: %v", err) + } + }) + + t.Run("cleanup enabled wipes the cache", func(t *testing.T) { + dir := t.TempDir() + p := New(Config{ + CacheDir: filepath.Join(dir, "cache"), ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", ContainerOS: "linux", Cleanup: true, Log: discardLogger(), + }) + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + get(t, "http://"+p.Addr()+"/index/se/rd/serde", nil) + if err := p.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "cache")); !os.IsNotExist(err) { + t.Errorf("cache survived cleanup=true (err=%v)", err) + } + }) +} + +// TestDLTemplateSurvivesRestart: Cargo caches config.json in CARGO_HOME, so +// after a daemon restart it may go straight to a crate download. The proxy +// must recover the upstream dl template from its own cache. +func TestDLTemplateSurvivesRestart(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/mirror/{crate}/{version}.crate"}`, "") + up.set("/mirror/serde/1.0.203.crate", "TARBALL", "") + + dir := t.TempDir() + mk := func() *Proxy { + return New(Config{ + CacheDir: filepath.Join(dir, "cache"), ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", ContainerOS: "linux", Log: discardLogger(), + }) + } + + p1 := mk() + if err := p1.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + get(t, "http://"+p1.Addr()+"/index/config.json", nil) + if err := p1.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + + p2 := mk() + if err := p2.Start(); err != nil { + t.Fatalf("restart: %v", err) + } + defer func() { _ = p2.Stop() }() + + if got := p2.dlTemplateOrDefault(); got != up.URL+"/mirror/{crate}/{version}.crate" { + t.Fatalf("dl template after restart = %q, want the cached upstream template", got) + } + resp, body := get(t, "http://"+p2.Addr()+"/crates/serde/1.0.203/download", nil) + if resp.StatusCode != http.StatusOK || body != "TARBALL" { + t.Errorf("status = %d body = %q, want the tarball via the recovered template", resp.StatusCode, body) + } +} + +func TestHealthz(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + if resp, _ := get(t, base+healthRoute, nil); resp.StatusCode != http.StatusOK { + t.Errorf("healthz status = %d, want 200", resp.StatusCode) + } +} + +func TestUnknownRouteIs404(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + if resp, _ := get(t, base+"/nope", nil); resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 404", resp.StatusCode) + } +} + +func TestNonGetMethodsRejected(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + req, err := http.NewRequest(http.MethodPost, base+"/index/se/rd/serde", nil) + if err != nil { + t.Fatalf("building request: %v", err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("status = %d, want 405", resp.StatusCode) + } +} diff --git a/pkg/proxies/go/goproxy.go b/pkg/proxies/go/goproxy.go index 4ba2e405..6fa639ed 100644 --- a/pkg/proxies/go/goproxy.go +++ b/pkg/proxies/go/goproxy.go @@ -28,10 +28,10 @@ import ( // Config for the Go module caching proxy. type Config struct { - CacheDir string // on-disk cache directory - Upstream string // upstream proxy URL (default: https://proxy.golang.org) - ListenAddr string // address to listen on (e.g., "10.88.0.1:8082") - Cleanup bool // wipe cache dir on Stop + CacheDir string // on-disk cache directory + Upstream string // upstream proxy URL (default: https://proxy.golang.org) + ListenAddr string // address to listen on (e.g., "10.88.0.1:8082") + Cleanup bool // wipe cache dir on Stop Log *slog.Logger } @@ -68,9 +68,13 @@ func (p *Proxy) Start() error { return fmt.Errorf("creating cache dir: %w", err) } - ln, err := net.Listen("tcp", p.cfg.ListenAddr) + // proxies.Listen (not net.Listen): the bridge gateway IP does not exist + // yet at daemon boot — CNI creates the ephemerd0 bridge with the first + // job container — so a direct bind fails with EADDRNOTAVAIL and the + // proxy silently never starts. See proxies.Listen for the full story. + ln, err := proxies.Listen(p.cfg.ListenAddr, p.cfg.Log) if err != nil { - return fmt.Errorf("listening on %s: %w", p.cfg.ListenAddr, err) + return err } p.listener = ln @@ -123,10 +127,28 @@ func (p *Proxy) Addr() string { } // EnvVars returns the environment variables to inject into job containers. +// +// The advertised address is the CONFIGURED one (the bridge gateway), not +// p.Addr(): when the listener falls back to the wildcard, Addr() reports +// "[::]:8082", which is meaningless inside a container. +// +// The "|" separator (not ",") is what makes this fail open. With ",direct" +// the go command only falls through to the origin on 404/410 — a proxy that +// is down, wedged, or returning 5xx hard-fails the build. With "|direct" it +// falls through on ANY error, so a broken cache degrades to a slower build +// instead of a red job. func (p *Proxy) EnvVars() []string { return []string{ - "GOPROXY=http://" + p.Addr() + ",direct", + "GOPROXY=http://" + p.advertiseAddr() + "|direct", + } +} + +// advertiseAddr is the address containers should use to reach this proxy. +func (p *Proxy) advertiseAddr() string { + if p.cfg.ListenAddr != "" { + return p.cfg.ListenAddr } + return p.Addr() } // Name returns the proxy name for logging. diff --git a/pkg/proxies/listen.go b/pkg/proxies/listen.go new file mode 100644 index 00000000..f19d8431 --- /dev/null +++ b/pkg/proxies/listen.go @@ -0,0 +1,79 @@ +package proxies + +import ( + "fmt" + "log/slog" + "net" +) + +// Listen binds a cache-proxy listener on addr, falling back to the wildcard +// address on the same port when addr itself cannot be bound. +// +// WHY THIS EXISTS (this is the bug that made the Go module proxy cache stay +// empty forever — see docs and the review notes on this change): +// +// Cache proxies are asked to listen on the CNI bridge gateway IP (e.g. +// 10.88.0.1:8082) so job containers can reach them. But ephemerd deletes the +// ephemerd0 bridge on shutdown and again on startup, and the CNI bridge +// plugin only (re)creates the bridge — and only then assigns the gateway IP +// to it — when the FIRST job container is networked. Proxies start long +// before that, during daemon boot, so net.Listen("tcp", "10.88.0.1:8082") +// fails with EADDRNOTAVAIL every single time. The caller logs a warning and +// carries on without the proxy, which means its env vars are never injected +// into any container: the cache directory is created and then never written +// to again. +// +// Binding the wildcard address fixes this without racing the bridge: a +// wildcard (INADDR_ANY) socket accepts connections addressed to interface +// addresses that appear AFTER the bind, so the gateway IP becomes reachable +// the moment CNI brings the bridge up. +// +// The requested address is always tried first, so a host where the gateway +// already exists keeps the narrower binding. When both fail (e.g. the port is +// already in use, which fails on the wildcard too) the ORIGINAL error is +// returned — the fallback never masks a real problem. +// +// SECURITY NOTE: the wildcard binding also exposes the proxy on the host's +// other interfaces. These proxies only ever serve public package-registry +// content and hold no credentials, so the exposure is limited to an open +// caching mirror. Operators who need it closed should firewall the port at +// the host edge; see docs/getting-started/configuration.md. +func Listen(addr string, log *slog.Logger) (net.Listener, error) { + ln, err := net.Listen("tcp", addr) + if err == nil { + return ln, nil + } + + wildcard, werr := wildcardAddr(addr) + if werr != nil { + return nil, fmt.Errorf("listening on %s: %w", addr, err) + } + + ln2, err2 := net.Listen("tcp", wildcard) + if err2 != nil { + // The wildcard failed too (port in use, permissions). Report the + // original failure — it is the one the operator configured for. + return nil, fmt.Errorf("listening on %s: %w", addr, err) + } + + if log != nil { + log.Info("cache proxy bound to wildcard address; the requested address is not on any interface yet (the CNI bridge is created with the first job container)", + "requested", addr, "bound", ln2.Addr().String(), "reason", err) + } + return ln2, nil +} + +// wildcardAddr rewrites a host:port address to the wildcard form ":port", +// which binds every interface. Returns an error for input that is not a +// valid host:port pair, so callers can fall back to reporting the original +// bind failure rather than listening somewhere unintended. +func wildcardAddr(addr string) (string, error) { + _, port, err := net.SplitHostPort(addr) + if err != nil { + return "", fmt.Errorf("splitting address %q: %w", addr, err) + } + if port == "" { + return "", fmt.Errorf("address %q has no port", addr) + } + return net.JoinHostPort("", port), nil +} diff --git a/pkg/proxies/listen_test.go b/pkg/proxies/listen_test.go new file mode 100644 index 00000000..a4fd7e23 --- /dev/null +++ b/pkg/proxies/listen_test.go @@ -0,0 +1,122 @@ +package proxies + +import ( + "io" + "log/slog" + "net" + "testing" +) + +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func TestWildcardAddr(t *testing.T) { + tests := []struct { + name string + addr string + want string + wantErr bool + }{ + {name: "gateway ipv4", addr: "10.88.0.1:8082", want: ":8082"}, + {name: "loopback", addr: "127.0.0.1:1", want: ":1"}, + {name: "already wildcard", addr: ":8083", want: ":8083"}, + {name: "zero port keeps zero", addr: "10.88.0.1:0", want: ":0"}, + {name: "ipv6 host", addr: "[fd00::1]:8082", want: ":8082"}, + {name: "hostname", addr: "gateway.internal:9000", want: ":9000"}, + {name: "no port", addr: "10.88.0.1", wantErr: true}, + {name: "empty", addr: "", wantErr: true}, + {name: "trailing colon only", addr: "10.88.0.1:", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := wildcardAddr(tt.addr) + if tt.wantErr { + if err == nil { + t.Fatalf("wildcardAddr(%q) = %q, want error", tt.addr, got) + } + return + } + if err != nil { + t.Fatalf("wildcardAddr(%q): %v", tt.addr, err) + } + if got != tt.want { + t.Errorf("wildcardAddr(%q) = %q, want %q", tt.addr, got, tt.want) + } + }) + } +} + +// TestListen_BindableAddressIsUsedDirectly pins that the fallback does not +// kick in when the requested address is actually assignable: loopback is +// always present, so the listener must end up on 127.0.0.1 and not on the +// wildcard. +func TestListen_BindableAddressIsUsedDirectly(t *testing.T) { + ln, err := Listen("127.0.0.1:0", testLogger()) + if err != nil { + t.Fatalf("Listen: %v", err) + } + defer func() { _ = ln.Close() }() + + host, _, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort(%q): %v", ln.Addr().String(), err) + } + if host != "127.0.0.1" { + t.Errorf("bound host = %q, want 127.0.0.1 (fallback should not trigger)", host) + } +} + +// TestListen_FallsBackToWildcard is the regression test for the empty-cache +// bug: binding an address that is not on any interface (the bridge gateway +// before CNI creates the bridge) must NOT fail the proxy, it must fall back +// to the wildcard so the address works once the interface appears. +// +// 203.0.113.0/24 is TEST-NET-3 (RFC 5737) and is never assigned to a real +// interface, so this reproduces EADDRNOTAVAIL without any network access. +func TestListen_FallsBackToWildcard(t *testing.T) { + ln, err := Listen("203.0.113.1:0", testLogger()) + if err != nil { + t.Fatalf("Listen fell over instead of falling back: %v", err) + } + defer func() { _ = ln.Close() }() + + host, port, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort(%q): %v", ln.Addr().String(), err) + } + if host == "203.0.113.1" { + t.Fatalf("bound the TEST-NET address %q — the test assumption is broken", ln.Addr()) + } + if port == "0" || port == "" { + t.Errorf("bound port = %q, want a real ephemeral port", port) + } + // A wildcard listener must be reachable over loopback. + c, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", port)) + if err != nil { + t.Fatalf("wildcard listener not reachable on loopback: %v", err) + } + _ = c.Close() +} + +// TestListen_ReturnsOriginalErrorWhenWildcardAlsoFails pins that the +// fallback never masks a genuine failure: when the port is already taken the +// wildcard bind fails too and the caller gets an error, not a silent success. +func TestListen_ReturnsOriginalErrorWhenWildcardAlsoFails(t *testing.T) { + // Occupy a port on the wildcard address so no further bind can succeed. + blocker, err := net.Listen("tcp", ":0") + if err != nil { + t.Fatalf("setting up blocker listener: %v", err) + } + defer func() { _ = blocker.Close() }() + _, port, err := net.SplitHostPort(blocker.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort: %v", err) + } + + ln, err := Listen(net.JoinHostPort("203.0.113.1", port), testLogger()) + if err == nil { + _ = ln.Close() + t.Skip("host allows rebinding an in-use wildcard port; cannot exercise this path here") + } +} diff --git a/pkg/proxies/proxy.go b/pkg/proxies/proxy.go index dbd5f242..bf20354a 100644 --- a/pkg/proxies/proxy.go +++ b/pkg/proxies/proxy.go @@ -6,6 +6,7 @@ // Implementations live in sub-packages: // // pkg/proxies/go/ — Go module proxy (GOPROXY) +// pkg/proxies/cargo/ — Cargo sparse registry + crate/rustup proxy // pkg/proxies/npm/ — (future) npm registry proxy // pkg/proxies/pip/ — (future) pip index proxy package proxies @@ -30,3 +31,28 @@ type CacheProxy interface { // Name returns a human-readable name for logging (e.g., "go", "npm"). Name() string } + +// Mount is a host→container bind mount a cache proxy needs in every job +// container. Env vars are enough for toolchains that take their proxy from +// the environment (Go, rustup); Cargo is not one of them — its source +// replacement is only read from a config file, never from CARGO_* env vars +// (verified empirically; see pkg/proxies/cargo). Such proxies generate the +// file on the host and declare it here. +type Mount struct { + // Source is an absolute host path (a directory). + Source string + // Destination is the absolute path inside the container. + Destination string + // ReadOnly mounts the source read-only. Config material should always + // be read-only: a job must never be able to rewrite what the next job + // on this host will read. + ReadOnly bool +} + +// MountProvider is an OPTIONAL interface a CacheProxy may implement when +// env vars alone cannot point a toolchain at the proxy. Callers type-assert +// for it; a proxy that does not implement it needs no mounts. +type MountProvider interface { + // Mounts returns the bind mounts to add to every job container spec. + Mounts() []Mount +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index b6bf8add..69024114 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -25,6 +25,7 @@ import ( "github.com/ephpm/ephemerd/pkg/config" "github.com/ephpm/ephemerd/pkg/dind" "github.com/ephpm/ephemerd/pkg/networking" + "github.com/ephpm/ephemerd/pkg/proxies" craneTarball "github.com/google/go-containerregistry/pkg/v1/tarball" ocispec "github.com/opencontainers/runtime-spec/specs-go" ) @@ -82,6 +83,12 @@ type Config struct { // config.DindConfig.AllowPrivileged for the threat model. DindAllowPrivileged bool CacheProxyEnv []string // extra env vars from cache proxies (e.g., GOPROXY=...) + // CacheProxyMounts are read-only bind mounts requested by cache proxies + // for toolchains that cannot be redirected with an env var. The Cargo + // proxy uses this to place a generated .cargo/config.toml at the + // container's filesystem root, where Cargo's ancestor-directory config + // search finds it for any workspace path. + CacheProxyMounts []proxies.Mount // Rlimits sets POSIX resource limits on each runner container's OCI // process. Zero values fall back to the containerd default (1024). // Applies on Linux only; ignored on Windows (HCS uses a different model). @@ -96,7 +103,7 @@ type Config struct { // construction site that forgets this field breaks `sudo` rather // than silently loosening the sandbox. AllowNewPrivileges bool - Network *networking.Manager + Network *networking.Manager // WindowsMemoryBytes is the memory limit for Hyper-V isolated Windows // runner containers. Zero leaves the OCI spec field unset, which gives // the HCS default (~1 GB) — too small for MSVC builds. Caller should @@ -666,6 +673,12 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err // Docker-in-Docker is not supported (no CAP_SYS_ADMIN/CAP_NET_ADMIN). oci.WithCapabilities(containerCapabilities), } + // Cache-proxy config mounts (e.g. the Cargo source-replacement config). + // Read-only: a job must never be able to rewrite what the next job on + // this host will read. + if len(r.cfg.CacheProxyMounts) > 0 { + opts = append(opts, withCacheProxyMounts(r.cfg.CacheProxyMounts)) + } opts = append(opts, seccompOpts()...) // AppArmor is an additional, independent layer over what the default spec // above already does (read-only /proc/sys and /sys, masked /proc paths, @@ -1276,6 +1289,44 @@ func withRunnerMount(hostDir, containerDir string) oci.SpecOpts { } } +// withCacheProxyMounts adds the bind mounts a cache proxy needs in every job +// container. The sources are host directories ephemerd generates and owns; +// runc creates the destination if the image does not have it. +func withCacheProxyMounts(mounts []proxies.Mount) oci.SpecOpts { + return func(_ context.Context, _ oci.Client, _ *containers.Container, s *oci.Spec) error { + for _, m := range mounts { + if m.Source == "" || m.Destination == "" { + continue + } + if goruntime.GOOS == "windows" { + opts := []string{"rw"} + if m.ReadOnly { + opts = []string{"ro"} + } + s.Mounts = append(s.Mounts, ocispec.Mount{ + Destination: m.Destination, + Source: m.Source, + Options: opts, + }) + continue + } + opts := []string{"rbind", "rw"} + if m.ReadOnly { + // "ro" alone is not enough on a recursive bind: without + // rprivate a later host-side mount could propagate in. + opts = []string{"rbind", "ro", "rprivate"} + } + s.Mounts = append(s.Mounts, ocispec.Mount{ + Destination: m.Destination, + Type: "bind", + Source: m.Source, + Options: opts, + }) + } + return nil + } +} + // copyDirForJob creates a writable copy of src at dst for a single job. // On Linux, uses hardlinks (cp -al) for instant, space-efficient copies. // On Windows, uses a native Go walk+copy — xcopy returned exit 4 (init From a8ea693139b0ad7f0f6a21ab729fd850a1275ccf Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 11 Aug 2026 19:14:18 -0700 Subject: [PATCH 2/2] ci(lint): exclude bodyclose from the cargo proxy tests Every request in cargoproxy_test.go goes through a get() helper that reads the body and closes it in a defer. bodyclose tracks the *http.Response the helper returns rather than what the helper does with it, so it reported all 30-odd call sites. The bodies are closed; restructuring the helper to avoid returning a response would churn 60 call sites to satisfy a false positive. Scoped to the test files only, matching the existing localtunnel exclusion. --- .golangci.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.golangci.yml b/.golangci.yml index 5d450009..8011474c 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -26,6 +26,13 @@ linters: linters: - errorlint - bodyclose + # The cargo proxy tests route every request through a get() helper that + # reads the body and closes it in a defer. bodyclose tracks the + # *http.Response the helper returns, not what the helper does with it, + # so it flags all 30-odd call sites. Same reasoning as localtunnel above. + - path: "pkg/proxies/cargo/.*_test\\.go" + linters: + - bodyclose - path: "api/v1" linters: - errorlint