diff --git a/cmd/ephemerd/actionscache.go b/cmd/ephemerd/actionscache.go new file mode 100644 index 0000000..f67e45e --- /dev/null +++ b/cmd/ephemerd/actionscache.go @@ -0,0 +1,75 @@ +package main + +import ( + "fmt" + "log/slog" + runtime_ "runtime" + + "github.com/ephpm/ephemerd/pkg/config" + "github.com/ephpm/ephemerd/pkg/networking" + "github.com/ephpm/ephemerd/pkg/providers" + "github.com/ephpm/ephemerd/pkg/proxies/actionscache" + "github.com/ephpm/ephemerd/pkg/scheduler" +) + +// actionsCachePorts returns the gateway port [actions_cache] binds, so the +// container firewall allows it. Linux only, like the cache itself. +func actionsCachePorts(cfg *config.Config) []int { + if !cfg.ActionsCache.Enabled || runtime_.GOOS != "linux" { + return nil + } + return []int{cfg.ActionsCache.ResolvedPort()} +} + +// startActionsCache starts the LAN Actions cache when [actions_cache] is +// enabled, returning the scheduler's handle on it (nil when off) and a +// teardown func. +// +// FAIL-OPEN: a cache that cannot start is logged and left out — jobs keep +// GitHub's cache. One that starts with its registry down is kept: it turns +// itself on for new jobs as soon as the registry answers (the scheduler asks +// Healthy() per job). +func startActionsCache(cfg *config.Config, net *networking.Manager, log *slog.Logger) (scheduler.ActionsCache, func()) { + noop := func() {} + ac := cfg.ActionsCache + if !ac.Enabled { + return nil, noop + } + if runtime_.GOOS != "linux" { + log.Warn("actions_cache is Linux-only; leaving it off on this host", "goos", runtime_.GOOS) + return nil, noop + } + addr := fmt.Sprintf("%s:%d", net.GatewayIP(), ac.ResolvedPort()) + p, err := actionscache.New(actionscache.Config{ + ListenAddr: addr, + AdvertiseBase: "http://" + addr, + Registry: ac.Registry, + RegistryUser: ac.Username, + RegistryPassword: ac.ResolvedPassword(), + MaxBytesPerRepo: ac.MaxBytesPerRepo(), + MaxEntryBytes: ac.MaxEntryBytes(), + UpstreamResultsURL: ac.UpstreamResultsURL, + JobForIP: net.JobForIP, + Log: log, + }) + if err != nil { + log.Warn("actions cache misconfigured; jobs use GitHub's cache", "error", err) + return nil, noop + } + if err := p.Start(); err != nil { + log.Warn("failed to start actions cache; jobs use GitHub's cache", "error", err) + return nil, noop + } + return actionsCacheAdapter{p}, func() { + if err := p.Stop(); err != nil { + log.Warn("error stopping actions cache", "error", err) + } + } +} + +// actionsCacheAdapter carries the provider's scope type into the proxy's. +type actionsCacheAdapter struct{ *actionscache.Proxy } + +func (a actionsCacheAdapter) Bind(runner string, s providers.CacheScope) { + a.Proxy.Bind(runner, actionscache.Scope{Repo: s.Repo, Write: s.Write, Reads: s.Reads}) +} diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 6997429..b46103f 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -478,6 +478,8 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP } // Language package caches (npm, pip, pub) — see pkgproxies.go. gatewayPorts = append(gatewayPorts, pkgProxyPorts(cfg)...) + // LAN Actions cache — see actionscache.go. + gatewayPorts = append(gatewayPorts, actionsCachePorts(cfg)...) // Initialize container networking, retrying through the cold-boot window. // @@ -586,6 +588,12 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP defer stopPkgProxies() cacheProxies = append(cacheProxies, pkgProxies...) + // The LAN Actions cache is per-job, not a CacheProxy: its env var is set + // only for jobs the scheduler turns it on for, so it is not collected + // into cacheProxyEnvVars below. + actionsCache, stopActionsCache := startActionsCache(cfg, net, log) + defer stopActionsCache() + // 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 @@ -873,6 +881,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP RunnerImageForRepo: cfg.Runner.ImageForRepoOS, Retry: retryCfg, OrphanSweep: orphanCfg, + ActionsCache: actionsCache, Log: log, }) diff --git a/pkg/config/actionscache.go b/pkg/config/actionscache.go new file mode 100644 index 0000000..cbfa7a3 --- /dev/null +++ b/pkg/config/actionscache.go @@ -0,0 +1,113 @@ +package config + +import ( + "errors" + "os" + "strings" +) + +// ActionsCacheConfig configures [actions_cache]: GitHub Actions' cache +// service (what actions/cache and Swatinem/rust-cache talk to) served from +// the LAN instead of GitHub, with entries stored in a writable OCI registry +// (registry:2 in normal, non-proxy mode — the pull-through mirror is +// read-only and cannot hold them). +// +// Linux hosts only. Off by default; with it off nothing about a job changes. +// With it on, a job only uses the LAN cache when the registry answers its +// health probe at job start AND the job's runner could be patched; otherwise +// it uses GitHub's cache exactly as before. +// +// Which entries a job may read and write is decided by ephemerd from +// GitHub's record of the job's run — branch pushes save under their branch, +// pull requests under their PR, and other events are read-only. See +// pkg/proxies/actionscache. +type ActionsCacheConfig struct { + // Enabled turns the cache on. Default false. + Enabled bool `toml:"enabled"` + // Port is the listen port on the bridge gateway. Default 8089. + Port int `toml:"port"` + + // Registry is the writable registry's base URL, e.g. + // "http://192.168.16.224:5001". Required when enabled. + Registry string `toml:"registry"` + // Username / Password authenticate to the registry (htpasswd). Jobs + // never see them: only ephemerd talks to the registry. + Username string `toml:"username"` + Password string `toml:"password"` + // PasswordEnv names an environment variable to read the password from + // instead of writing it in this file. Wins over Password when set. + PasswordEnv string `toml:"password_env"` + + // MaxSizeGBPerRepo caps each repository's total cache; the least + // recently used entries are deleted past it. Default 100. 0 keeps the + // default; a negative value disables eviction. + MaxSizeGBPerRepo int `toml:"max_size_gb_per_repo"` + // MaxEntryGB refuses a single entry larger than this. Default 10. + MaxEntryGB int `toml:"max_entry_gb"` + + // UpstreamResultsURL is where requests that are not the cache (artifact + // uploads and downloads) are relayed. Default + // "https://results-receiver.actions.githubusercontent.com". + UpstreamResultsURL string `toml:"upstream_results_url"` +} + +// DefaultActionsCachePort is the gateway port [actions_cache] binds when +// port is unset — the next free one after the package caches (8082-8088). +const DefaultActionsCachePort = 8089 + +// ResolvedPort returns Port or the default. +func (c ActionsCacheConfig) ResolvedPort() int { + if c.Port > 0 { + return c.Port + } + return DefaultActionsCachePort +} + +// ResolvedPassword returns the registry password, from PasswordEnv when set. +func (c ActionsCacheConfig) ResolvedPassword() string { + if c.PasswordEnv != "" { + return os.Getenv(c.PasswordEnv) + } + return c.Password +} + +// MaxBytesPerRepo returns the eviction cap in bytes (0 = no eviction). +func (c ActionsCacheConfig) MaxBytesPerRepo() int64 { + switch { + case c.MaxSizeGBPerRepo < 0: + return 0 + case c.MaxSizeGBPerRepo == 0: + return 100 << 30 + default: + return int64(c.MaxSizeGBPerRepo) << 30 + } +} + +// MaxEntryBytes returns the per-entry limit in bytes. +func (c ActionsCacheConfig) MaxEntryBytes() int64 { + if c.MaxEntryGB <= 0 { + return 10 << 30 + } + return int64(c.MaxEntryGB) << 30 +} + +func (c ActionsCacheConfig) validate() error { + if !c.Enabled { + return nil + } + if strings.TrimSpace(c.Registry) == "" { + return errors.New("actions_cache.registry is required when actions_cache is enabled: the writable registry entries are stored in") + } + if err := validateMirrorURL("actions_cache.registry", c.Registry); err != nil { + return err + } + if c.UpstreamResultsURL != "" { + if err := validateMirrorURL("actions_cache.upstream_results_url", c.UpstreamResultsURL); err != nil { + return err + } + } + if c.PasswordEnv != "" && os.Getenv(c.PasswordEnv) == "" { + return errors.New("actions_cache.password_env names " + c.PasswordEnv + ", which is not set in ephemerd's environment") + } + return nil +} diff --git a/pkg/config/actionscache_test.go b/pkg/config/actionscache_test.go new file mode 100644 index 0000000..7833acf --- /dev/null +++ b/pkg/config/actionscache_test.go @@ -0,0 +1,33 @@ +package config + +import "testing" + +func TestActionsCacheConfig(t *testing.T) { + var off ActionsCacheConfig + if err := off.validate(); err != nil { + t.Fatalf("disabled block failed validation: %v", err) + } + if off.ResolvedPort() != DefaultActionsCachePort || off.MaxBytesPerRepo() != 100<<30 || off.MaxEntryBytes() != 10<<30 { + t.Fatal("defaults are wrong") + } + if err := (ActionsCacheConfig{Enabled: true}).validate(); err == nil { + t.Fatal("enabled without a registry passed validation") + } + if err := (ActionsCacheConfig{Enabled: true, Registry: "cache.lan:5001"}).validate(); err == nil { + t.Fatal("a scheme-less registry passed validation") + } + if err := (ActionsCacheConfig{Enabled: true, Registry: "http://cache.lan:5001"}).validate(); err != nil { + t.Fatalf("valid block failed: %v", err) + } + t.Setenv("EPHEMERD_TEST_AC_PASS", "hunter2") + c := ActionsCacheConfig{Enabled: true, Registry: "http://c:5001", Password: "file", PasswordEnv: "EPHEMERD_TEST_AC_PASS"} + if c.ResolvedPassword() != "hunter2" { + t.Fatal("password_env did not win over password") + } + if err := (ActionsCacheConfig{Enabled: true, Registry: "http://c:5001", PasswordEnv: "EPHEMERD_TEST_AC_UNSET"}).validate(); err == nil { + t.Fatal("password_env naming an unset variable passed validation") + } + if (ActionsCacheConfig{MaxSizeGBPerRepo: -1}).MaxBytesPerRepo() != 0 { + t.Fatal("negative size cap should disable eviction") + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 75f5c27..1a72e50 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -45,11 +45,14 @@ type Config struct { // RegistryMirror routes container image pulls through a LAN pull-through // cache instead of the origin registry. See RegistryMirrorConfig. RegistryMirror RegistryMirrorConfig `toml:"registry_mirror"` - Runtime RuntimeConfig `toml:"runtime"` - Runner RunnerConfig `toml:"runner"` - Metrics MetricsConfig `toml:"metrics"` - Dispatch DispatchConfig `toml:"dispatch"` - Log LogConfig `toml:"log"` + // ActionsCache serves GitHub Actions' cache from the LAN, backed by a + // writable registry. See ActionsCacheConfig. + ActionsCache ActionsCacheConfig `toml:"actions_cache"` + Runtime RuntimeConfig `toml:"runtime"` + Runner RunnerConfig `toml:"runner"` + Metrics MetricsConfig `toml:"metrics"` + Dispatch DispatchConfig `toml:"dispatch"` + Log LogConfig `toml:"log"` } // DispatchConfig configures the host<->VM dispatch gRPC channel. On Windows @@ -1984,6 +1987,10 @@ func (c *Config) validate() error { return err } + if err := c.ActionsCache.validate(); err != nil { + return err + } + if err := c.Runner.Linux.validate(); err != nil { return err } diff --git a/pkg/dind/containers.go b/pkg/dind/containers.go index 94ad577..c7fb5b9 100644 --- a/pkg/dind/containers.go +++ b/pkg/dind/containers.go @@ -490,7 +490,7 @@ func (s *Server) handleContainerCreate(w http.ResponseWriter, r *http.Request) { } if len(req.Env) > 0 { - opts = append(opts, oci.WithEnv(req.Env)) + opts = append(opts, oci.WithEnv(translateResultsEnv(req.Env, s.resultsURL))) } if req.WorkingDir != "" { diff --git a/pkg/dind/dind.go b/pkg/dind/dind.go index 50b7017..3c7665d 100644 --- a/pkg/dind/dind.go +++ b/pkg/dind/dind.go @@ -48,6 +48,7 @@ type Server struct { jobID string jobNamespace string // per-job containerd namespace for isolation cacheNamespace string // per-(provider,repo) shared image cache namespace; empty disables caching + resultsURL string // ACTIONS_RESULTS_URL for container steps; see resultsenv.go transport Transport // how the API is exposed to the job container; see listen.go dockerDir string // /jobs//docker — per-job scratch, exists on every transport sockPath string // host-side unix socket path; empty on the TCP transport @@ -198,6 +199,11 @@ type Config struct { // See listen.go. Transport Transport + // ResultsURL is the ACTIONS_RESULTS_URL the job's steps must see: the + // LAN Actions cache when it is on for this job, "" otherwise. See + // resultsenv.go for why dind has to restore it at all. + ResultsURL string + Log *slog.Logger } @@ -240,6 +246,7 @@ func New(cfg Config) (*Server, error) { allowPrivileged: cfg.AllowPrivileged, stager: newBindStager(cfg.DataDir, cfg.JobID, cfg.Log), mirror: cfg.RegistryMirror, + resultsURL: cfg.ResultsURL, log: cfg.Log.With("component", "dind", "job_id", cfg.JobID), images: make(map[string]*imageEntry), containers: make(map[string]*containerEntry), diff --git a/pkg/dind/exec.go b/pkg/dind/exec.go index b6f7390..737d48d 100644 --- a/pkg/dind/exec.go +++ b/pkg/dind/exec.go @@ -125,6 +125,7 @@ func (s *Server) handleExecCreate(w http.ResponseWriter, r *http.Request, contai env = append(env, spec.Process.Env...) } env = append(env, req.Env...) + env = translateResultsEnv(env, s.resultsURL) exec := &execEntry{ ID: execID, diff --git a/pkg/dind/resultsenv.go b/pkg/dind/resultsenv.go new file mode 100644 index 0000000..a068c6e --- /dev/null +++ b/pkg/dind/resultsenv.go @@ -0,0 +1,49 @@ +package dind + +import "strings" + +// Names the runner exports the results service address under: the real one, +// and the one a runner patched for the LAN Actions cache exports instead +// (pkg/runner.PatchResultsURL). Duplicated as literals so dind does not +// depend on the runner package. +const ( + resultsURLVar = "ACTIONS_RESULTS_URL=" + resultsRenamedVar = "ACTIONS_RESULTS_ORL=" +) + +// translateResultsEnv fixes ACTIONS_RESULTS_URL for a `container:` job. +// +// Host steps inherit the runner process's environment, which carries the +// ACTIONS_RESULTS_URL ephemerd set for the LAN cache. Container steps do +// not: the runner hands them only the step environment, through +// `docker exec -e` — and a patched runner exports GitHub's address there as +// ACTIONS_RESULTS_ORL. Left alone, a container step would see no +// ACTIONS_RESULTS_URL at all, and upload-artifact would fail the job. +// +// So whenever the renamed variable is present, ACTIONS_RESULTS_URL is set to +// the LAN cache (resultsURL) when this job has one, or restored to GitHub's +// address when it does not. Env without the renamed variable — every job not +// using the LAN cache — is returned untouched. +func translateResultsEnv(env []string, resultsURL string) []string { + original, found := "", false + for _, kv := range env { + if v, ok := strings.CutPrefix(kv, resultsRenamedVar); ok { + original, found = v, true + } + } + if !found { + return env + } + want := resultsURL + if want == "" { + want = original + } + out := make([]string, 0, len(env)+1) + for _, kv := range env { + if strings.HasPrefix(kv, resultsURLVar) { + continue + } + out = append(out, kv) + } + return append(out, resultsURLVar+want) +} diff --git a/pkg/dind/resultsenv_test.go b/pkg/dind/resultsenv_test.go new file mode 100644 index 0000000..698b431 --- /dev/null +++ b/pkg/dind/resultsenv_test.go @@ -0,0 +1,43 @@ +package dind + +import ( + "strings" + "testing" +) + +func TestTranslateResultsEnv(t *testing.T) { + const gh = "https://results-receiver.actions.githubusercontent.com/" + const lan = "http://10.88.0.1:8089/" + for _, tc := range []struct { + name string + env []string + resultsURL string + want string // value of ACTIONS_RESULTS_URL, "" = must be absent + untouched bool + }{ + {"not a patched runner: untouched", []string{"A=1", resultsURLVar + gh}, lan, gh, true}, + {"patched, LAN cache on", []string{"A=1", resultsRenamedVar + gh}, lan, lan, false}, + {"patched, cache off: GitHub's address restored", []string{resultsRenamedVar + gh}, "", gh, false}, + {"patched, stale URL replaced", []string{resultsURLVar + "http://stale/", resultsRenamedVar + gh}, lan, lan, false}, + {"no results vars at all", []string{"A=1"}, lan, "", true}, + } { + t.Run(tc.name, func(t *testing.T) { + got := translateResultsEnv(append([]string(nil), tc.env...), tc.resultsURL) + if tc.untouched && strings.Join(got, "\n") != strings.Join(tc.env, "\n") { + t.Fatalf("env changed: %v -> %v", tc.env, got) + } + var vals []string + for _, kv := range got { + if v, ok := strings.CutPrefix(kv, resultsURLVar); ok { + vals = append(vals, v) + } + } + switch { + case tc.want == "" && len(vals) != 0: + t.Fatalf("ACTIONS_RESULTS_URL = %v, want absent", vals) + case tc.want != "" && (len(vals) != 1 || vals[0] != tc.want): + t.Fatalf("ACTIONS_RESULTS_URL = %v, want exactly [%s]", vals, tc.want) + } + }) + } +} diff --git a/pkg/github/api_test.go b/pkg/github/api_test.go index d04e30a..e8bba4e 100644 --- a/pkg/github/api_test.go +++ b/pkg/github/api_test.go @@ -1244,3 +1244,62 @@ func TestIsSelfHosted_EdgeCases(t *testing.T) { }) } } + +// TestPollJobs_ReportsRunningJobsAndNeedsJobs covers what polling used to +// miss: a run is in_progress once its first job starts, so (a) a `needs:` +// job still queued inside it was never found, and (b) the running job was +// never reported, so the scheduler could not learn which job a runner took +// (and could not bind its LAN Actions cache scope). A run listed under both +// statuses must be read once. +func TestPollJobs_ReportsRunningJobsAndNeedsJobs(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/repos/testorg/repo1/actions/runs", func(w http.ResponseWriter, r *http.Request) { + runs := []map[string]any{} + switch r.URL.Query().Get("status") { + case "in_progress": + runs = append(runs, map[string]any{"id": 300, "status": "in_progress"}) + case "queued": + runs = append(runs, map[string]any{"id": 300, "status": "queued"}) // same run, both lists + } + if err := json.NewEncoder(w).Encode(map[string]any{"total_count": len(runs), "workflow_runs": runs}); err != nil { + t.Logf("encoding: %v", err) + } + }) + mux.HandleFunc("/repos/testorg/repo1/actions/runs/300/jobs", func(w http.ResponseWriter, r *http.Request) { + resp := map[string]any{ + "total_count": 3, + "jobs": []map[string]any{ + {"id": 1, "status": "in_progress", "runner_name": "ephemerd-a", "labels": []string{"self-hosted", "linux"}}, + {"id": 2, "status": "queued", "labels": []string{"self-hosted", "linux"}}, + {"id": 3, "status": "in_progress", "runner_name": "GitHub Actions 7", "labels": []string{"ubuntu-latest"}}, + }, + } + if err := json.NewEncoder(w).Encode(resp); err != nil { + t.Logf("encoding: %v", err) + } + }) + + c, srv := newTestClientWithServer(t, mux) + defer srv.Close() + + events, err := c.PollJobs(context.Background()) + if err != nil { + t.Fatalf("PollJobs() error: %v", err) + } + got := map[int64]string{} + for _, e := range events { + if _, dup := got[e.Job.GetID()]; dup { + t.Errorf("job %d reported twice; the run was listed under both statuses and must be read once", e.Job.GetID()) + } + got[e.Job.GetID()] = e.Action + } + if got[1] != "in_progress" { + t.Errorf("running self-hosted job: action %q, want in_progress", got[1]) + } + if got[2] != "queued" { + t.Errorf("needs-job queued inside an in_progress run: action %q, want queued", got[2]) + } + if _, ok := got[3]; ok { + t.Errorf("GitHub-hosted job must not be reported") + } +} diff --git a/pkg/github/cachefacts.go b/pkg/github/cachefacts.go new file mode 100644 index 0000000..398cf9a --- /dev/null +++ b/pkg/github/cachefacts.go @@ -0,0 +1,79 @@ +package github + +import ( + "context" + "fmt" + "time" +) + +// RunCacheFacts is the part of a workflow run that decides which LAN cache +// scope its jobs get. Everything here comes from GitHub's API, not from the +// job. +type RunCacheFacts struct { + // Repo is the run's repository as "owner/repo". + Repo string + // Event is what triggered the run ("push", "pull_request", …). + Event string + // HeadBranch is the branch (or, for a tag push, the tag) the run is for. + HeadBranch string + // HeadRepo is the repository the head commit lives in, "owner/repo". + // It differs from Repo for a pull request from a fork. + HeadRepo string + // PRNumber is the pull request the run belongs to, when GitHub lists + // one (it does not for fork PRs). 0 otherwise. + PRNumber int + // DefaultBranch is the repository's default branch. + DefaultBranch string + RunID int64 +} + +type defaultBranchEntry struct { + branch string + fetched time.Time +} + +const defaultBranchTTL = time.Hour + +// RunCacheFacts fetches the facts for one run of repo (a name under this +// client's owner). It costs one API call per run, plus one per repository +// per hour for the default branch, which the run payload does not reliably +// carry. +func (c *Client) RunCacheFacts(ctx context.Context, repo string, runID int64) (RunCacheFacts, error) { + run, _, err := c.client.Actions.GetWorkflowRunByID(ctx, c.cfg.Owner, repo, runID) + if err != nil { + return RunCacheFacts{}, fmt.Errorf("fetching run %d of %s/%s: %w", runID, c.cfg.Owner, repo, err) + } + f := RunCacheFacts{ + Repo: run.GetRepository().GetFullName(), + Event: run.GetEvent(), + HeadBranch: run.GetHeadBranch(), + HeadRepo: run.GetHeadRepository().GetFullName(), + RunID: runID, + } + if f.Repo == "" { + f.Repo = c.cfg.Owner + "/" + repo + } + for _, pr := range run.PullRequests { + if n := pr.GetNumber(); n > 0 { + f.PRNumber = n + break + } + } + f.DefaultBranch = run.GetRepository().GetDefaultBranch() + if f.DefaultBranch == "" { + if v, ok := c.defaultBranches.Load(repo); ok { + if e := v.(defaultBranchEntry); time.Since(e.fetched) < defaultBranchTTL { + f.DefaultBranch = e.branch + } + } + } + if f.DefaultBranch == "" { + r, _, err := c.client.Repositories.Get(ctx, c.cfg.Owner, repo) + if err != nil { + return RunCacheFacts{}, fmt.Errorf("fetching default branch of %s/%s: %w", c.cfg.Owner, repo, err) + } + f.DefaultBranch = r.GetDefaultBranch() + c.defaultBranches.Store(repo, defaultBranchEntry{branch: f.DefaultBranch, fetched: time.Now()}) + } + return f, nil +} diff --git a/pkg/github/client.go b/pkg/github/client.go index bf64371..b2b3a1f 100644 --- a/pkg/github/client.go +++ b/pkg/github/client.go @@ -11,6 +11,7 @@ import ( "log/slog" "net/http" "strconv" + "sync" "sync/atomic" "time" @@ -53,6 +54,8 @@ type Client struct { client *gh.Client app *AppAuth // nil when using PAT lastRate *rateSnapshot // last observed rate-limit state (nil in tests using SetHTTPClient) + + defaultBranches sync.Map // repo name -> defaultBranchEntry (see RunCacheFacts) } // RateSnapshot returns the last-observed rate-limit state. The scheduler @@ -489,29 +492,56 @@ func (c *Client) pollOrg(ctx context.Context) ([]JobEvent, error) { return events, nil } -// pollRepo checks a single repo for queued workflow jobs. +// pollRepo checks a single repo for workflow jobs and reports them the way +// webhooks would: "queued" for self-hosted jobs waiting on a runner, and +// "in_progress" for self-hosted jobs a runner has taken. +// +// Runs are listed in both the queued and the in_progress state. A run is +// in_progress as soon as its first job starts, so a later job waiting on +// `needs:` sits queued inside an in_progress run — listing queued runs alone +// never finds it. And the in_progress events are what tell the scheduler +// which job a runner actually took; without them polling mode could not +// bind a runner's LAN Actions cache scope, so every save was refused. func (c *Client) pollRepo(ctx context.Context, repo string) ([]JobEvent, error) { var events []JobEvent - runs, _, err := c.client.Actions.ListRepositoryWorkflowRuns(ctx, c.cfg.Owner, repo, &gh.ListWorkflowRunsOptions{ - Status: "queued", - }) - if err != nil { - return nil, err + seen := map[int64]bool{} + var runs []*gh.WorkflowRun + for i, status := range []string{"queued", "in_progress"} { + page, _, err := c.client.Actions.ListRepositoryWorkflowRuns(ctx, c.cfg.Owner, repo, &gh.ListWorkflowRunsOptions{ + Status: status, + }) + if err != nil { + if i == 0 { + return nil, err + } + // The queued listing already worked; losing the in_progress + // half costs only latency, so keep what we have. + c.cfg.Log.Debug("listing in-progress runs failed", "repo", repo, "error", err) + continue + } + for _, run := range page.WorkflowRuns { + if !seen[run.GetID()] { + seen[run.GetID()] = true + runs = append(runs, run) + } + } } - for _, run := range runs.WorkflowRuns { + for _, run := range runs { runJobs, _, err := c.client.Actions.ListWorkflowJobs(ctx, c.cfg.Owner, repo, run.GetID(), nil) if err != nil { continue } for _, job := range runJobs.Jobs { - if job.GetStatus() == "queued" && isSelfHosted(job.Labels) { - events = append(events, JobEvent{ - Action: "queued", - Repo: repo, - Job: job, - }) + if !isSelfHosted(job.Labels) { + continue + } + switch { + case job.GetStatus() == "queued": + events = append(events, JobEvent{Action: "queued", Repo: repo, Job: job}) + case job.GetStatus() == "in_progress" && job.GetRunnerName() != "": + events = append(events, JobEvent{Action: "in_progress", Repo: repo, Job: job}) } } } diff --git a/pkg/networking/firewall_linux.go b/pkg/networking/firewall_linux.go index ab4d8d1..da83bf6 100644 --- a/pkg/networking/firewall_linux.go +++ b/pkg/networking/firewall_linux.go @@ -497,6 +497,23 @@ func (l *linuxNetworking) joinJobNetwork(cniID, jobID, containerIP string) error return nil } +// jobForIP returns the job whose member holds ip, or "". +func (l *linuxNetworking) jobForIP(ip string) string { + parsed := net.ParseIP(strings.TrimSpace(ip)) + if parsed == nil { + return "" + } + want := parsed.String() + l.c2cMu.Lock() + defer l.c2cMu.Unlock() + for _, m := range l.c2cMember { + if m.ip == want { + return m.jobID + } + } + return "" +} + // leaveJobNetwork removes the container attached under cniID from its job and // deletes the intra-job ACCEPT pairs it shared with every remaining member. // Best-effort on the iptables side: a rule that fails to delete is logged and diff --git a/pkg/networking/firewall_linux_test.go b/pkg/networking/firewall_linux_test.go index 97f8c81..db5fc45 100644 --- a/pkg/networking/firewall_linux_test.go +++ b/pkg/networking/firewall_linux_test.go @@ -531,6 +531,34 @@ func TestJoinJobNetwork_SkipsUnusableIP(t *testing.T) { l.leaveJobNetwork("never-joined") } +// TestJobForIP maps a source address back to its job — the attribution the +// LAN Actions cache relies on — and forgets it once the container leaves. +// Membership is seeded directly so no iptables is needed. +func TestJobForIP(t *testing.T) { + l := &linuxNetworking{cfg: Config{Subnet: "10.88.0.0/16", Log: testLogger()}} + l.c2cMember = map[string]c2cMember{ + "runner-a": {jobID: "runner-a", ip: "10.88.0.7"}, + "sibling-a": {jobID: "runner-a", ip: "10.88.0.8"}, + "runner-b": {jobID: "runner-b", ip: "10.88.0.9"}, + } + l.c2cJob = map[string]map[string]bool{ + "runner-a": {"10.88.0.7": true, "10.88.0.8": true}, + "runner-b": {"10.88.0.9": true}, + } + for ip, want := range map[string]string{ + "10.88.0.7": "runner-a", "10.88.0.8": "runner-a", " 10.88.0.9 ": "runner-b", + "10.88.0.99": "", "garbage": "", "": "", + } { + if got := l.jobForIP(ip); got != want { + t.Errorf("jobForIP(%q) = %q, want %q", ip, got, want) + } + } + delete(l.c2cMember, "sibling-a") // what leaveJobNetwork does first + if got := l.jobForIP("10.88.0.8"); got != "" { + t.Errorf("a departed container still maps to %q", got) + } +} + // The port is the whole scope on the deny side, so a caller passing an // address the CNI result did not produce must fail rather than open // something wider or something wrong. diff --git a/pkg/networking/firewall_windows.go b/pkg/networking/firewall_windows.go index 94c068e..2292fb7 100644 --- a/pkg/networking/firewall_windows.go +++ b/pkg/networking/firewall_windows.go @@ -467,6 +467,7 @@ func (w *windowsNetworking) closeHostPort(port int, containerIP string) { // scope per job at this layer. No-op. func (w *windowsNetworking) joinJobNetwork(_, _, _ string) error { return nil } func (w *windowsNetworking) leaveJobNetwork(_ string) {} +func (w *windowsNetworking) jobForIP(_ string) string { return "" } // hostPortRulePrefix is the DisplayName prefix of every per-job host-port allow // (see hostPortAllowRule). Swept by prefix on shutdown so a hard kill — which diff --git a/pkg/networking/network_darwin.go b/pkg/networking/network_darwin.go index 922b672..c84f03a 100644 --- a/pkg/networking/network_darwin.go +++ b/pkg/networking/network_darwin.go @@ -65,3 +65,4 @@ func (d *darwinNetworking) closeHostPort(int, string) {} // has no container bridge of its own. func (d *darwinNetworking) joinJobNetwork(_, _, _ string) error { return nil } func (d *darwinNetworking) leaveJobNetwork(_ string) {} +func (d *darwinNetworking) jobForIP(_ string) string { return "" } diff --git a/pkg/networking/networking.go b/pkg/networking/networking.go index 4797add..42133aa 100644 --- a/pkg/networking/networking.go +++ b/pkg/networking/networking.go @@ -253,6 +253,10 @@ type platformNetworking interface { // so both are no-ops. joinJobNetwork(cniID, jobID, containerIP string) error leaveJobNetwork(cniID string) + + // jobForIP returns the jobID a container at ip joined under, or "" when + // no current member has that address. Backs JobForIP. + jobForIP(ip string) string } // New creates and initializes the networking manager for the current platform. @@ -380,6 +384,23 @@ func (m *Manager) LeaveJobNetwork(cniID string) { } } +// JobForIP reports which job the container at ip belongs to — the jobID it +// joined under via JoinJobNetwork (the runner container and every dind +// sibling of one job share it) — or "" when no current member has that +// address. +// +// It lets a host-side service tell which job a request came from by its +// source address alone. That holds on the CNI bridge because each container +// owns its address inside its own netns and job containers lack CAP_NET_RAW, +// so one cannot complete a TCP handshake as another. Always "" on Windows +// and macOS, which do not track membership. +func (m *Manager) JobForIP(ip string) string { + if m == nil || m.platform == nil { + return "" + } + return m.platform.jobForIP(ip) +} + // Cleanup removes all networking state: firewall rules, bridge interface, // CNI config, IP allocations, and DNS files. Called on shutdown. func (m *Manager) Cleanup() { diff --git a/pkg/providers/github/cachescope.go b/pkg/providers/github/cachescope.go new file mode 100644 index 0000000..a40538d --- /dev/null +++ b/pkg/providers/github/cachescope.go @@ -0,0 +1,84 @@ +package github + +import ( + "context" + "errors" + "strconv" + "strings" + + "github.com/ephpm/ephemerd/pkg/github" + "github.com/ephpm/ephemerd/pkg/providers" +) + +var _ providers.CacheScoper = (*Provider)(nil) + +// CacheScope resolves the LAN cache scope for the job in event from GitHub's +// record of its run. +func (p *Provider) CacheScope(ctx context.Context, event *providers.JobEvent) (providers.CacheScope, error) { + if event == nil || event.RunID == 0 || event.Repo == "" { + return providers.CacheScope{}, errors.New("event carries no run to resolve a cache scope from") + } + f, err := p.client.RunCacheFacts(ctx, event.Repo, event.RunID) + if err != nil { + return providers.CacheScope{}, err + } + return cacheScopeFor(f), nil +} + +// trustedPushEvents run code that is already on a branch of the repository, +// so a job may save under that branch — exactly as GitHub's own cache +// allows. +var trustedPushEvents = map[string]bool{ + "push": true, + "schedule": true, + "workflow_dispatch": true, + "repository_dispatch": true, + "release": true, + "create": true, + "merge_group": true, +} + +// cacheScopeFor is the policy, kept pure so it can be tested in full. +// +// - Branch events (push, schedule, dispatch, …) save under their own +// branch and restore from it, then the default branch. +// - pull_request saves under the PR's merge ref and restores from it, then +// the default branch; nothing a PR saves is ever visible to a branch. +// A fork PR is treated the same way — its writes are confined to its own +// PR scope — and when GitHub does not name the PR (it omits fork PRs from +// the run payload) the scope is the run itself, so it is shared with +// nothing. +// - Everything else (pull_request_target, issue_comment, workflow_run, …) +// is READ-ONLY. GitHub would let some of these save under the default +// branch, but they are the events that run on behalf of outsiders, and +// a poisoned default-branch entry is restored by every job in the repo. +// A miss costs minutes; a poisoned cache costs the build. +// - A head repository that differs from the run's repository on any +// event other than pull_request is read-only too: that combination does +// not occur legitimately. +func cacheScopeFor(f github.RunCacheFacts) providers.CacheScope { + s := providers.CacheScope{Repo: f.Repo} + def := "" + if f.DefaultBranch != "" { + def = "refs/heads/" + f.DefaultBranch + } + fork := f.HeadRepo != "" && !strings.EqualFold(f.HeadRepo, f.Repo) + + switch { + case f.Event == "pull_request": + own := "run/" + strconv.FormatInt(f.RunID, 10) + if f.PRNumber > 0 { + own = "refs/pull/" + strconv.Itoa(f.PRNumber) + "/merge" + } + s.Write = own + s.Reads = []string{own} + case trustedPushEvents[f.Event] && !fork && f.HeadBranch != "": + own := "refs/heads/" + f.HeadBranch + s.Write = own + s.Reads = []string{own} + } + if def != "" && (len(s.Reads) == 0 || s.Reads[0] != def) { + s.Reads = append(s.Reads, def) + } + return s +} diff --git a/pkg/providers/github/cachescope_test.go b/pkg/providers/github/cachescope_test.go new file mode 100644 index 0000000..1a736d9 --- /dev/null +++ b/pkg/providers/github/cachescope_test.go @@ -0,0 +1,62 @@ +package github + +import ( + "strings" + "testing" + + "github.com/ephpm/ephemerd/pkg/github" +) + +func TestCacheScopeFor(t *testing.T) { + base := github.RunCacheFacts{Repo: "ephpm/ephpm", HeadRepo: "ephpm/ephpm", DefaultBranch: "main", RunID: 42} + with := func(mut func(*github.RunCacheFacts)) github.RunCacheFacts { + f := base + mut(&f) + return f + } + for _, tc := range []struct { + name string + f github.RunCacheFacts + write string + reads string + }{ + {"push to main", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch = "push", "main" }), + "refs/heads/main", "refs/heads/main"}, + {"push to a feature branch", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch = "push", "feat/x" }), + "refs/heads/feat/x", "refs/heads/feat/x,refs/heads/main"}, + {"schedule", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch = "schedule", "main" }), + "refs/heads/main", "refs/heads/main"}, + {"same-repo PR", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch, f.PRNumber = "pull_request", "feat/x", 7 }), + "refs/pull/7/merge", "refs/pull/7/merge,refs/heads/main"}, + // The poisoning case: a fork names its branch "main". It must land in + // its own PR/run scope, never in refs/heads/main. + {"fork PR with a branch called main", with(func(f *github.RunCacheFacts) { + f.Event, f.HeadBranch, f.HeadRepo = "pull_request", "main", "attacker/ephpm" + }), "run/42", "run/42,refs/heads/main"}, + {"pull_request_target is read-only", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch = "pull_request_target", "main" }), + "", "refs/heads/main"}, + {"issue_comment is read-only", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch = "issue_comment", "main" }), + "", "refs/heads/main"}, + {"push from a foreign head repo is read-only", with(func(f *github.RunCacheFacts) { + f.Event, f.HeadBranch, f.HeadRepo = "push", "main", "attacker/ephpm" + }), "", "refs/heads/main"}, + {"unknown default branch", with(func(f *github.RunCacheFacts) { f.Event, f.HeadBranch, f.DefaultBranch = "push", "dev", "" }), + "refs/heads/dev", "refs/heads/dev"}, + } { + t.Run(tc.name, func(t *testing.T) { + s := cacheScopeFor(tc.f) + if s.Repo != "ephpm/ephpm" { + t.Errorf("Repo = %q", s.Repo) + } + if s.Write != tc.write { + t.Errorf("Write = %q, want %q", s.Write, tc.write) + } + if got := strings.Join(s.Reads, ","); got != tc.reads { + t.Errorf("Reads = %q, want %q", got, tc.reads) + } + if s.Write != "" && s.Write == "refs/heads/main" && tc.f.Event != "push" && tc.f.Event != "schedule" { + t.Errorf("event %q got write access to the default branch", tc.f.Event) + } + }) + } +} diff --git a/pkg/providers/provider.go b/pkg/providers/provider.go index 3d32c45..6a9dc62 100644 --- a/pkg/providers/provider.go +++ b/pkg/providers/provider.go @@ -122,6 +122,32 @@ type RunnerNameReporter interface { ReportsRunnerNames() bool } +// CacheScope is what one job may do with ephemerd's LAN Actions cache. +// Mirrors actionscache.Scope; kept here so providers need not import the +// proxy. +type CacheScope struct { + // Repo is the forge-native "owner/repo". + Repo string + // Write is the ref the job saves entries under; "" = read-only. + Write string + // Reads are the refs a restore searches, in order. + Reads []string +} + +// CacheScoper is optionally implemented by providers that can say, from +// their own record of a job's run, which cache scope that job gets. +// +// It is asked for the job that ACTUALLY took a runner (the in_progress +// event), never the one the runner was dispatched for: same-label JIT +// runners are fungible, so dispatch intent says nothing about whose code +// will run. An error means "unknown" and the job gets no cache access — +// a miss, never a wider scope. +type CacheScoper interface { + Provider + + CacheScope(ctx context.Context, event *JobEvent) (CacheScope, error) +} + // RunnerBusyReporter is optionally implemented by providers that can be // asked, on demand, whether a runner they registered is executing a job // right now. diff --git a/pkg/proxies/actionscache/actionscache.go b/pkg/proxies/actionscache/actionscache.go new file mode 100644 index 0000000..76ddf6d --- /dev/null +++ b/pkg/proxies/actionscache/actionscache.go @@ -0,0 +1,976 @@ +// Package actionscache serves GitHub Actions' cache service (v2) from the +// LAN, backed by a writable OCI registry, so `actions/cache` and everything +// built on it (Swatinem/rust-cache, setup-* caching) stop moving +// gigabyte-sized archives over the WAN on every job. +// +// How a job reaches it. The runner exports ACTIONS_RESULTS_URL to every step +// from the job message, overwriting the container's environment, so ephemerd +// patches that name out of the job's own copy of Runner.Worker.dll (see +// pkg/runner.PatchResultsURL) and sets ACTIONS_RESULTS_URL to this server. +// The cache client then talks to us; GitHub's real address survives under a +// renamed variable and is what we relay everything that is not the cache to. +// +// Who a request is from. The cache client builds its twirp URLs with an +// absolute path ("/twirp/..."), which discards any path we could have put a +// per-job secret in, and it sends GitHub's own runtime token, which we cannot +// verify. So a request is attributed by its SOURCE ADDRESS: every container +// of a job joins that job's network under the job's id (networking.JobForIP), +// and job containers cannot spoof another's address on the bridge. The upload +// and download URLs we hand out are used verbatim, so those additionally carry +// an unguessable token. +// +// What a job may touch. Not decided here: a job is unbound until the +// scheduler binds it to a Scope derived from the forge's record of the job +// that ACTUALLY took the runner (GitHub hands same-label JIT runners to any +// queued job, so the job a runner was started for is not evidence). Until +// then every lookup misses and every save is refused. +// +// Failure is always a cache miss. Unbound, registry down, entry missing: +// the twirp answer is ok=false (which the client treats as "no cache" or +// "could not reserve", warning and moving on), never a 5xx, which it would +// retry five times and then log as an error. +package actionscache + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "encoding/xml" + "errors" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "net/http/httputil" + "net/url" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/ephpm/ephemerd/pkg/proxies" +) + +// Config configures the cache server. +type Config struct { + // ListenAddr is the address to bind, normally the bridge gateway + // ("10.88.0.1:8089"). + ListenAddr string + // AdvertiseBase is the base URL job containers reach this server at + // ("http://10.88.0.1:8089"). Upload and download URLs are built on it. + AdvertiseBase string + + // Registry is the writable registry's base URL ("http://cache:5001"). + Registry string + RegistryUser string + RegistryPassword string + + // MaxBytesPerRepo caps each forge repo's total cache; the least + // recently used entries are deleted past it. 0 disables eviction. + MaxBytesPerRepo int64 + // MaxEntryBytes refuses a single save larger than this. Default 10 GiB + // (GitHub's own per-repo limit). + MaxEntryBytes int64 + + // UpstreamResultsURL is GitHub's results service, where every request + // that is not the cache (artifact uploads, …) is relayed unchanged. + // Default https://results-receiver.actions.githubusercontent.com. + UpstreamResultsURL string + + // JobForIP attributes a request to a job by its source address + // (networking.Manager.JobForIP). Required. + JobForIP func(ip string) string + + // HealthInterval is how often the registry is probed; negative disables + // the loop (tests). Default 30s. + HealthInterval time.Duration + + // HTTPClient talks to the registry. Default: a client without a total + // timeout (blobs are large) but with dial/response-header bounds. + HTTPClient *http.Client + + Log *slog.Logger +} + +const ( + defaultUpstreamResultsURL = "https://results-receiver.actions.githubusercontent.com" + defaultMaxEntryBytes = 10 << 30 + defaultHealthInterval = 30 * time.Second + + twirpPrefix = "/twirp/github.actions.results.api.v1.CacheService/" + blobPrefix = "/_ephemerd/actions-cache/" + + pendingTTL = 2 * time.Hour + downloadTTL = time.Hour + sweepEvery = 10 * time.Minute +) + +// jobState is what the server knows about one runner. +type jobState struct { + bound bool + scope Scope +} + +// pending is a save between CreateCacheEntry and FinalizeCacheEntryUpload. +type pending struct { + runner string + scope Scope + key string + version string + created time.Time + + mu sync.Mutex + blocks map[string]descriptor // Azure block id -> uploaded layer + order []string // committed block list ("" = single-shot) + size int64 + failed error +} + +// download is a granted restore: the entry snapshot it may stream. +type download struct { + runner string + e *entry + expires time.Time +} + +// Proxy is the cache server. +type Proxy struct { + cfg Config + log *slog.Logger + reg *registry + ix *index + upstream *url.URL + relay *httputil.ReverseProxy + + healthy atomic.Bool + + mu sync.Mutex + jobs map[string]*jobState // runner name -> state + uploads map[string]*pending // upload token -> pending + byKey map[string]*pending // runner+key+version -> pending + downloads map[string]*download // download token -> grant + + srv *http.Server + ln net.Listener + cancel context.CancelFunc + wg sync.WaitGroup +} + +// New validates cfg and builds the server. It does not listen until Start. +func New(cfg Config) (*Proxy, error) { + if cfg.JobForIP == nil { + return nil, errors.New("actionscache: JobForIP is required") + } + if cfg.Log == nil { + cfg.Log = slog.Default() + } + if cfg.MaxEntryBytes <= 0 { + cfg.MaxEntryBytes = defaultMaxEntryBytes + } + if cfg.HealthInterval == 0 { + cfg.HealthInterval = defaultHealthInterval + } + if cfg.UpstreamResultsURL == "" { + cfg.UpstreamResultsURL = defaultUpstreamResultsURL + } + if cfg.HTTPClient == nil { + cfg.HTTPClient = &http.Client{Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext, + ResponseHeaderTimeout: 2 * time.Minute, + MaxIdleConnsPerHost: 16, + IdleConnTimeout: 90 * time.Second, + }} + } + cfg.AdvertiseBase = strings.TrimRight(cfg.AdvertiseBase, "/") + if _, err := url.Parse(cfg.AdvertiseBase); err != nil || cfg.AdvertiseBase == "" { + return nil, fmt.Errorf("actionscache: advertise base %q is not a URL", cfg.AdvertiseBase) + } + reg, err := newRegistry(cfg.Registry, cfg.RegistryUser, cfg.RegistryPassword, cfg.HTTPClient) + if err != nil { + return nil, fmt.Errorf("actionscache: %w", err) + } + up, err := url.Parse(strings.TrimRight(cfg.UpstreamResultsURL, "/")) + if err != nil || up.Scheme == "" || up.Host == "" { + return nil, fmt.Errorf("actionscache: upstream results URL %q is not an absolute URL", cfg.UpstreamResultsURL) + } + p := &Proxy{ + cfg: cfg, + log: cfg.Log.With("component", "actions-cache"), + reg: reg, + ix: newIndex(reg), + upstream: up, + jobs: map[string]*jobState{}, + uploads: map[string]*pending{}, + byKey: map[string]*pending{}, + downloads: map[string]*download{}, + } + p.relay = &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + pr.SetURL(up) + pr.Out.Host = up.Host + }, + ErrorLog: slog.NewLogLogger(p.log.Handler(), slog.LevelWarn), + } + return p, nil +} + +// Start probes the registry once, binds the listener and starts serving. +func (p *Proxy) Start() error { + ctx, cancel := context.WithCancel(context.Background()) + p.cancel = cancel + p.probe(ctx) + + // proxies.Listen, not net.Listen: the gateway address only exists once + // the first job container brings the CNI bridge up, which is after the + // daemon starts this server — a plain bind fails with EADDRNOTAVAIL on + // every fresh start. The wildcard fallback is safe here because the + // server attributes every request to a job by source IP and refuses the + // rest; the advertised URL is AdvertiseBase, never the bound address. + ln, err := proxies.Listen(p.cfg.ListenAddr, p.log) + if err != nil { + cancel() + return fmt.Errorf("actionscache: listen %s: %w", p.cfg.ListenAddr, err) + } + p.ln = ln + p.srv = &http.Server{Handler: p, ReadHeaderTimeout: 30 * time.Second} + p.wg.Add(1) + go func() { + defer p.wg.Done() + if err := p.srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + p.log.Error("actions cache server stopped", "error", err) + } + }() + if p.cfg.HealthInterval > 0 { + p.wg.Add(1) + go func() { + defer p.wg.Done() + t := time.NewTicker(p.cfg.HealthInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + p.probe(ctx) + p.expire(time.Now()) + } + } + }() + } + p.log.Info("actions cache ready", "addr", ln.Addr().String(), "registry", p.reg.base.String(), "healthy", p.Healthy()) + return nil +} + +// Stop shuts the server down. +func (p *Proxy) Stop() error { + if p.cancel != nil { + p.cancel() + } + var err error + if p.srv != nil { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + err = p.srv.Shutdown(ctx) + cancel() + } + p.wg.Wait() + return err +} + +// Addr is the bound address (after Start). +func (p *Proxy) Addr() string { + if p.ln == nil { + return p.cfg.ListenAddr + } + return p.ln.Addr().String() +} + +// ResultsURL is the ACTIONS_RESULTS_URL value for job containers. +func (p *Proxy) ResultsURL() string { return p.cfg.AdvertiseBase + "/" } + +// Healthy reports whether the registry answered the last probe. The +// scheduler only turns the cache on for a new job while this is true. +func (p *Proxy) Healthy() bool { return p.healthy.Load() } + +func (p *Proxy) probe(ctx context.Context) { + pctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + err := p.reg.ping(pctx) + was := p.healthy.Swap(err == nil) + switch { + case err != nil && was: + p.log.Warn("actions cache registry stopped answering; new jobs use GitHub's cache", "error", err) + case err == nil && !was: + p.log.Info("actions cache registry is answering") + } +} + +// Track registers a runner the cache is turned on for. Until Bind it is +// unbound: every lookup misses, every save is refused. +func (p *Proxy) Track(runner string) { + p.mu.Lock() + defer p.mu.Unlock() + if _, ok := p.jobs[runner]; !ok { + p.jobs[runner] = &jobState{} + } +} + +// Bind grants a tracked runner its scope. It is a no-op for an untracked +// runner (the cache was not turned on for it) and refuses to re-bind: a +// runner runs exactly one job, so a second, different binding is a bug or an +// attack, and the safe answer is to drop the runner's access entirely. +func (p *Proxy) Bind(runner string, s Scope) { + p.mu.Lock() + defer p.mu.Unlock() + st, ok := p.jobs[runner] + if !ok { + return + } + if st.bound { + if !sameScope(st.scope, s) { + p.log.Warn("conflicting second cache binding for a runner; revoking its cache access", "runner", runner, "first", st.scope.Repo+"@"+st.scope.Write, "second", s.Repo+"@"+s.Write) + st.scope = Scope{} + } + return + } + st.bound, st.scope = true, s + p.log.Info("cache scope bound", "runner", runner, "repo", s.Repo, "write", s.Write, "reads", strings.Join(s.Reads, ",")) +} + +func sameScope(a, b Scope) bool { + return a.Repo == b.Repo && a.Write == b.Write && strings.Join(a.Reads, "\n") == strings.Join(b.Reads, "\n") +} + +// Forget drops a runner and everything it was granted. Call it when the +// runner's environment is destroyed. +func (p *Proxy) Forget(runner string) { + p.mu.Lock() + defer p.mu.Unlock() + delete(p.jobs, runner) + for tok, u := range p.uploads { + if u.runner == runner { + delete(p.uploads, tok) + delete(p.byKey, runner+"\x00"+u.key+"\x00"+u.version) + } + } + for tok, d := range p.downloads { + if d.runner == runner { + delete(p.downloads, tok) + } + } +} + +// expire drops grants nobody finished with. +func (p *Proxy) expire(now time.Time) { + p.mu.Lock() + defer p.mu.Unlock() + for tok, u := range p.uploads { + if now.Sub(u.created) > pendingTTL { + delete(p.uploads, tok) + delete(p.byKey, u.runner+"\x00"+u.key+"\x00"+u.version) + } + } + for tok, d := range p.downloads { + if now.After(d.expires) { + delete(p.downloads, tok) + } + } +} + +// caller identifies the job a request came from: its runner name and, if +// bound, its scope. +func (p *Proxy) caller(r *http.Request) (runner string, s Scope, bound bool) { + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + return "", Scope{}, false + } + runner = p.cfg.JobForIP(host) + if runner == "" { + return "", Scope{}, false + } + p.mu.Lock() + defer p.mu.Unlock() + st, ok := p.jobs[runner] + if !ok { + return "", Scope{}, false + } + return runner, st.scope, st.bound && st.scope.Repo != "" +} + +func token() string { + var b [24]byte + if _, err := rand.Read(b[:]); err != nil { + panic(err) // crypto/rand does not fail on supported platforms + } + return hex.EncodeToString(b[:]) +} + +// ServeHTTP routes cache calls here and relays everything else to GitHub. +func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasPrefix(r.URL.Path, twirpPrefix): + p.serveTwirp(w, r, strings.TrimPrefix(r.URL.Path, twirpPrefix)) + case strings.HasPrefix(r.URL.Path, blobPrefix+"u/"): + p.serveUpload(w, r, strings.TrimPrefix(r.URL.Path, blobPrefix+"u/")) + case strings.HasPrefix(r.URL.Path, blobPrefix+"d/"): + p.serveDownload(w, r, strings.TrimPrefix(r.URL.Path, blobPrefix+"d/")) + default: + p.serveRelay(w, r) + } +} + +// serveRelay passes a non-cache request (artifacts, …) to GitHub untouched, +// Authorization included. Only containers of a job the cache is on for may +// use it — this must not become an open relay on the bridge. +func (p *Proxy) serveRelay(w http.ResponseWriter, r *http.Request) { + host, _, _ := net.SplitHostPort(r.RemoteAddr) + runner := p.cfg.JobForIP(host) + p.mu.Lock() + _, tracked := p.jobs[runner] + p.mu.Unlock() + if runner == "" || !tracked { + http.Error(w, "not a job this cache serves", http.StatusForbidden) + return + } + p.relay.ServeHTTP(w, r) +} + +// --- twirp --------------------------------------------------------------- + +// The cache client (actions/toolkit packages/cache, generated +// results/api/v1/cache.ts) serializes requests with useProtoFieldName, so +// field names arrive snake_case and int64 values as JSON strings; it parses +// responses leniently (either naming, string or number). Every field below +// was checked against that generated code; none is guessed. + +type twirpReq struct { + Key string `json:"key"` + Version string `json:"version"` + RestoreKeys []string `json:"restore_keys"` + SizeBytes jsonInt `json:"size_bytes"` +} + +// jsonInt accepts an int64 as a JSON number or string (proto3 JSON emits +// strings). +type jsonInt int64 + +func (j *jsonInt) UnmarshalJSON(b []byte) error { + s := strings.Trim(string(b), `"`) + if s == "" || s == "null" { + *j = 0 + return nil + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil { + return err + } + *j = jsonInt(n) + return nil +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +// twirpError answers in twirp's error envelope. Only used for requests that +// are malformed or for methods we do not serve; a 4xx is not retried. +func twirpError(w http.ResponseWriter, status int, code, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]string{"code": code, "msg": msg}) +} + +func (p *Proxy) serveTwirp(w http.ResponseWriter, r *http.Request, method string) { + if r.Method != http.MethodPost { + twirpError(w, http.StatusBadRequest, "bad_route", "twirp methods are POST") + return + } + var req twirpReq + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil { + twirpError(w, http.StatusBadRequest, "malformed", err.Error()) + return + } + runner, scope, bound := p.caller(r) + switch method { + case "GetCacheEntryDownloadURL": + p.getDownloadURL(w, r, runner, scope, bound, req) + case "CreateCacheEntry": + p.createEntry(w, r, runner, scope, bound, req) + case "FinalizeCacheEntryUpload": + p.finalizeEntry(w, r, runner, scope, bound, req) + default: + twirpError(w, http.StatusNotFound, "bad_route", "method not served by ephemerd's LAN cache: "+method) + } +} + +func (p *Proxy) getDownloadURL(w http.ResponseWriter, r *http.Request, runner string, s Scope, bound bool, req twirpReq) { + miss := map[string]any{"ok": false, "signed_download_url": "", "matched_key": ""} + if !bound || !p.Healthy() { + writeJSON(w, miss) + return + } + e, err := p.ix.lookup(r.Context(), s, req.Key, req.RestoreKeys, req.Version) + if err != nil { + p.log.Warn("cache lookup failed; answering as a miss", "runner", runner, "error", err) + } + if e == nil { + writeJSON(w, miss) + return + } + tok := token() + now := time.Now() + p.mu.Lock() + p.downloads[tok] = &download{runner: runner, e: e, expires: now.Add(downloadTTL)} + p.mu.Unlock() + p.touch(e, now) + p.log.Info("cache hit", "runner", runner, "repo", s.Repo, "key", e.Key, "ref", e.Ref, "bytes", e.Size) + writeJSON(w, map[string]any{ + "ok": true, + "signed_download_url": p.cfg.AdvertiseBase + blobPrefix + "d/" + tok, + "matched_key": e.Key, + }) +} + +func (p *Proxy) touch(e *entry, now time.Time) { + s := p.ix.scope(e.repository) + s.mu.Lock() + e.lastUsed = now + s.mu.Unlock() +} + +func (p *Proxy) createEntry(w http.ResponseWriter, r *http.Request, runner string, s Scope, bound bool, req twirpReq) { + refuse := func(msg string) { + writeJSON(w, map[string]any{"ok": false, "signed_upload_url": "", "message": msg}) + } + switch { + case !bound: + refuse("ephemerd's LAN cache has no scope for this job yet; not saving") + return + case s.Write == "": + refuse("this job may only restore from the cache, not save to it") + return + case !p.Healthy(): + refuse("ephemerd's LAN cache registry is not answering; not saving") + return + case req.Key == "" || req.Version == "": + refuse("key and version are required") + return + } + exists, err := p.ix.exists(r.Context(), s.Repo, s.Write, req.Key, req.Version) + if err != nil { + refuse("checking for an existing entry failed: " + err.Error()) + return + } + if exists { + refuse("an entry for this key already exists; cache entries are immutable") + return + } + tok := token() + u := &pending{runner: runner, scope: s, key: req.Key, version: req.Version, created: time.Now(), blocks: map[string]descriptor{}} + bk := runner + "\x00" + req.Key + "\x00" + req.Version + p.mu.Lock() + if prev, ok := p.byKey[bk]; ok { + for t, v := range p.uploads { + if v == prev { + delete(p.uploads, t) + } + } + } + p.uploads[tok] = u + p.byKey[bk] = u + p.mu.Unlock() + writeJSON(w, map[string]any{ + "ok": true, + "signed_upload_url": p.cfg.AdvertiseBase + blobPrefix + "u/" + tok, + "message": "", + }) +} + +func (p *Proxy) finalizeEntry(w http.ResponseWriter, r *http.Request, runner string, s Scope, bound bool, req twirpReq) { + refuse := func(msg string) { + writeJSON(w, map[string]any{"ok": false, "entry_id": "0", "message": msg}) + } + if !bound || s.Write == "" { + refuse("this job may not save to the cache") + return + } + bk := runner + "\x00" + req.Key + "\x00" + req.Version + p.mu.Lock() + u, ok := p.byKey[bk] + if ok { + delete(p.byKey, bk) + for t, v := range p.uploads { + if v == u { + delete(p.uploads, t) + } + } + } + p.mu.Unlock() + if !ok { + refuse("no upload in progress for this key") + return + } + + u.mu.Lock() + layers, size, err := u.committed() + u.mu.Unlock() + if err != nil { + refuse(err.Error()) + return + } + if int64(req.SizeBytes) != size { + refuse(fmt.Sprintf("uploaded %d bytes but the client says the archive is %d", size, req.SizeBytes)) + return + } + + e := &entry{ + Key: u.key, Version: u.version, Ref: u.scope.Write, + Created: time.Now().UTC(), Size: size, Layers: layers, + repository: scopeRepository(u.scope.Repo, u.scope.Write), + tag: entryTag(u.key, u.version), + } + cfgBytes := configBlob(e) + cfgDigest, _, err := p.reg.uploadBlob(r.Context(), e.repository, strings.NewReader(string(cfgBytes))) + if err != nil { + refuse("storing entry config: " + err.Error()) + return + } + body, err := encodeManifest(e, descriptor{MediaType: configMediaType, Digest: cfgDigest, Size: int64(len(cfgBytes))}) + if err != nil { + refuse(err.Error()) + return + } + digest, err := p.reg.putManifest(r.Context(), e.repository, e.tag, body) + if err != nil { + refuse("storing entry manifest: " + err.Error()) + return + } + e.digest, e.lastUsed = digest, e.Created + p.ix.add(e) + p.log.Info("cache saved", "runner", runner, "repo", u.scope.Repo, "ref", e.Ref, "key", e.Key, "bytes", e.Size, "blocks", len(layers)) + + if evicted, err := p.ix.evict(r.Context(), u.scope.Repo, p.cfg.MaxBytesPerRepo, sweepEvery, time.Now()); err != nil { + p.log.Warn("cache eviction failed", "repo", u.scope.Repo, "error", err) + } else if len(evicted) > 0 { + var freed int64 + for _, ev := range evicted { + freed += ev.Size + } + p.log.Info("cache evicted least recently used entries", "repo", u.scope.Repo, "entries", len(evicted), "bytes", freed) + } + + writeJSON(w, map[string]any{"ok": true, "entry_id": strconv.FormatInt(e.Created.UnixNano(), 10), "message": ""}) +} + +// committed returns the upload's layers in block-list order. Caller holds +// u.mu. +func (u *pending) committed() ([]descriptor, int64, error) { + if u.failed != nil { + return nil, 0, fmt.Errorf("upload failed: %w", u.failed) + } + if u.order == nil { + return nil, 0, errors.New("upload was never committed") + } + layers := make([]descriptor, 0, len(u.order)) + var size int64 + for _, id := range u.order { + d, ok := u.blocks[id] + if !ok { + return nil, 0, fmt.Errorf("block list names block %q that was never uploaded", id) + } + layers = append(layers, d) + size += d.Size + } + return layers, size, nil +} + +// --- Azure blob upload --------------------------------------------------- + +// The client uploads with @azure/storage-blob's BlockBlobClient.uploadFile +// (toolkit uploadUtils.ts uploadCacheArchiveSDK): one Put Blob for archives +// up to 128 MiB, otherwise concurrent Put Block calls and a Put Block List. +// Each block becomes its own registry blob, so out-of-order concurrent blocks +// stream straight through without being reassembled first. + +func (p *Proxy) serveUpload(w http.ResponseWriter, r *http.Request, tok string) { + if r.Method != http.MethodPut { + http.Error(w, "uploads are PUT", http.StatusMethodNotAllowed) + return + } + p.mu.Lock() + u, ok := p.uploads[tok] + p.mu.Unlock() + runner, _, _ := p.caller(r) + if !ok || runner == "" || runner != u.runner { + http.Error(w, "unknown upload", http.StatusNotFound) + return + } + + q := r.URL.Query() + switch q.Get("comp") { + case "block": + id := q.Get("blockid") + if id == "" { + http.Error(w, "blockid is required", http.StatusBadRequest) + return + } + d, err := p.storeBlock(r, u) + if err != nil { + p.uploadFailed(w, u, err) + return + } + u.mu.Lock() + u.blocks[id] = d + u.mu.Unlock() + case "blocklist": + ids, err := parseBlockList(io.LimitReader(r.Body, 8<<20)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + u.mu.Lock() + u.order = ids + u.mu.Unlock() + case "": + // Put Blob: the whole archive in one request. + d, err := p.storeBlock(r, u) + if err != nil { + p.uploadFailed(w, u, err) + return + } + u.mu.Lock() + u.blocks[""] = d + u.order = []string{""} + u.mu.Unlock() + default: + http.Error(w, "unsupported comp="+q.Get("comp"), http.StatusBadRequest) + return + } + azureCreated(w) +} + +// storeBlock streams one request body into the registry, enforcing the +// per-entry size limit across all of the upload's blocks. +func (p *Proxy) storeBlock(r *http.Request, u *pending) (descriptor, error) { + u.mu.Lock() + room := p.cfg.MaxEntryBytes - u.size + u.mu.Unlock() + if room <= 0 { + return descriptor{}, fmt.Errorf("entry exceeds the %d byte limit", p.cfg.MaxEntryBytes) + } + lr := &io.LimitedReader{R: r.Body, N: room + 1} + digest, n, err := p.reg.uploadBlob(r.Context(), scopeRepository(u.scope.Repo, u.scope.Write), lr) + if err != nil { + return descriptor{}, err + } + if n > room { + return descriptor{}, fmt.Errorf("entry exceeds the %d byte limit", p.cfg.MaxEntryBytes) + } + u.mu.Lock() + u.size += n + u.mu.Unlock() + return descriptor{MediaType: chunkMediaType, Digest: digest, Size: n}, nil +} + +func (p *Proxy) uploadFailed(w http.ResponseWriter, u *pending, err error) { + u.mu.Lock() + if u.failed == nil { + u.failed = err + } + u.mu.Unlock() + p.log.Warn("cache upload failed; the job saves nothing", "runner", u.runner, "key", u.key, "error", err) + // 400, not 5xx: the Azure SDK retries 5xx, and the upload is already + // doomed — Finalize will refuse it. + http.Error(w, err.Error(), http.StatusBadRequest) +} + +// azureCreated answers a Put Blob / Put Block / Put Block List the way the +// Azure SDK expects to see success. +func azureCreated(w http.ResponseWriter) { + h := w.Header() + h.Set("ETag", `"0x`+strings.ToUpper(token()[:16])+`"`) + h.Set("Last-Modified", time.Now().UTC().Format(http.TimeFormat)) + h.Set("x-ms-request-id", token()[:32]) + h.Set("x-ms-version", "2024-11-04") + h.Set("x-ms-request-server-encrypted", "false") + w.WriteHeader(http.StatusCreated) +} + +// parseBlockList reads a Put Block List body: with , +// or children, in commit order. The ids are base64 +// as sent with Put Block and are compared verbatim. +func parseBlockList(r io.Reader) ([]string, error) { + dec := xml.NewDecoder(r) + var ids []string + inList := false + for { + tok, err := dec.Token() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return nil, fmt.Errorf("parsing block list: %w", err) + } + se, ok := tok.(xml.StartElement) + if !ok { + continue + } + switch se.Name.Local { + case "BlockList": + inList = true + case "Latest", "Committed", "Uncommitted": + if !inList { + return nil, errors.New("block id outside ") + } + var id string + if err := dec.DecodeElement(&id, &se); err != nil { + return nil, fmt.Errorf("parsing block id: %w", err) + } + id = strings.TrimSpace(id) + if _, err := base64.StdEncoding.DecodeString(id); err != nil { + return nil, fmt.Errorf("block id %q is not base64", id) + } + ids = append(ids, id) + } + } + if !inList { + return nil, errors.New("no element") + } + if len(ids) == 0 { + return nil, errors.New("empty block list") + } + return ids, nil +} + +// --- download ------------------------------------------------------------ + +// serveDownload streams an entry's layers back as one archive. The client +// (downloadCacheHttpClient, used for every non-Azure host) does a plain GET +// and checks the bytes against Content-Length; a single byte range is also +// honoured for robustness. +func (p *Proxy) serveDownload(w http.ResponseWriter, r *http.Request, tok string) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + http.Error(w, "downloads are GET", http.StatusMethodNotAllowed) + return + } + p.mu.Lock() + d, ok := p.downloads[tok] + p.mu.Unlock() + runner, _, _ := p.caller(r) + if !ok || runner == "" || runner != d.runner || time.Now().After(d.expires) { + http.Error(w, "unknown download", http.StatusNotFound) + return + } + e := d.e + + start, end := int64(0), e.Size-1 + partial := false + if rh := r.Header.Get("Range"); rh != "" && e.Size > 0 { + s, en, ok := parseRange(rh, e.Size) + if !ok { + w.Header().Set("Content-Range", "bytes */"+strconv.FormatInt(e.Size, 10)) + http.Error(w, "unsatisfiable range", http.StatusRequestedRangeNotSatisfiable) + return + } + start, end, partial = s, en, true + } + length := end - start + 1 + if e.Size == 0 { + length = 0 + } + h := w.Header() + h.Set("Content-Type", "application/octet-stream") + h.Set("Accept-Ranges", "bytes") + h.Set("Content-Length", strconv.FormatInt(length, 10)) + if partial { + h.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, e.Size)) + w.WriteHeader(http.StatusPartialContent) + } else { + w.WriteHeader(http.StatusOK) + } + if r.Method == http.MethodHead || length == 0 { + return + } + if err := p.streamRange(r.Context(), w, e, start, length); err != nil { + // Headers are gone; cutting the connection short makes the client's + // Content-Length check fail, which it treats as a failed restore + // (a miss), not a corrupt one. + p.log.Warn("cache download aborted", "runner", runner, "key", e.Key, "error", err) + panic(http.ErrAbortHandler) + } +} + +func (p *Proxy) streamRange(ctx context.Context, w io.Writer, e *entry, off, n int64) error { + for _, l := range e.Layers { + if n <= 0 { + return nil + } + if off >= l.Size { + off -= l.Size + continue + } + take := l.Size - off + if take > n { + take = n + } + rc, err := p.reg.openBlob(ctx, e.repository, l.Digest, off, take) + if err != nil { + return err + } + copied, err := io.Copy(w, rc) + _ = rc.Close() + if err != nil { + return err + } + if copied != take { + return fmt.Errorf("blob %s: got %d bytes, want %d", l.Digest, copied, take) + } + n -= take + off = 0 + } + if n > 0 { + return fmt.Errorf("entry shorter than its recorded size by %d bytes", n) + } + return nil +} + +// parseRange accepts one "bytes=a-b", "bytes=a-" or "bytes=-n" range. +func parseRange(h string, size int64) (start, end int64, ok bool) { + spec, found := strings.CutPrefix(strings.TrimSpace(h), "bytes=") + if !found || strings.Contains(spec, ",") { + return 0, 0, false + } + a, b, found := strings.Cut(spec, "-") + if !found { + return 0, 0, false + } + if a == "" { + n, err := strconv.ParseInt(b, 10, 64) + if err != nil || n <= 0 { + return 0, 0, false + } + if n > size { + n = size + } + return size - n, size - 1, true + } + s, err := strconv.ParseInt(a, 10, 64) + if err != nil || s < 0 || s >= size { + return 0, 0, false + } + e := size - 1 + if b != "" { + if e, err = strconv.ParseInt(b, 10, 64); err != nil || e < s { + return 0, 0, false + } + if e >= size { + e = size - 1 + } + } + return s, e, true +} diff --git a/pkg/proxies/actionscache/actionscache_test.go b/pkg/proxies/actionscache/actionscache_test.go new file mode 100644 index 0000000..7a5f3f1 --- /dev/null +++ b/pkg/proxies/actionscache/actionscache_test.go @@ -0,0 +1,595 @@ +package actionscache + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +// harness wires a Proxy to a fake registry. Every request in a test comes +// from 127.0.0.1, so ip maps it to whichever runner the test says. +type harness struct { + t *testing.T + reg *fakeRegistry + p *Proxy + srv *httptest.Server + mu sync.Mutex + runner string // runner 127.0.0.1 is attributed to +} + +func newHarness(t *testing.T, mutate ...func(*Config)) *harness { + t.Helper() + h := &harness{t: t, runner: "runner-1"} + h.reg, _ = newFakeRegistry(t) + regSrv := httptest.NewServer(h.reg) + t.Cleanup(regSrv.Close) + + cfg := Config{ + ListenAddr: "127.0.0.1:0", + AdvertiseBase: "http://placeholder", + Registry: regSrv.URL, + RegistryUser: "cache", + RegistryPassword: "s3cret", + HealthInterval: -1, + JobForIP: func(ip string) string { + h.mu.Lock() + defer h.mu.Unlock() + if ip == "127.0.0.1" { + return h.runner + } + return "" + }, + } + for _, m := range mutate { + m(&cfg) + } + p, err := New(cfg) + if err != nil { + t.Fatal(err) + } + h.p = p + h.srv = httptest.NewServer(p) + t.Cleanup(h.srv.Close) + p.cfg.AdvertiseBase = h.srv.URL + p.probe(context.Background()) + if !p.Healthy() { + t.Fatal("proxy not healthy against the fake registry") + } + return h +} + +func (h *harness) as(runner string) { + h.mu.Lock() + h.runner = runner + h.mu.Unlock() +} + +func (h *harness) twirp(method string, req map[string]any) map[string]any { + h.t.Helper() + b, _ := json.Marshal(req) + resp, err := http.Post(h.srv.URL+twirpPrefix+method, "application/json", bytes.NewReader(b)) + if err != nil { + h.t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + var out map[string]any + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + h.t.Fatalf("%s: decoding response (status %s): %v", method, resp.Status, err) + } + if resp.StatusCode != http.StatusOK { + h.t.Fatalf("%s: status %s, body %v (cache failures must be ok=false, not an HTTP error)", method, resp.Status, out) + } + return out +} + +func put(t *testing.T, url string, body []byte) int { + t.Helper() + req, _ := http.NewRequest(http.MethodPut, url, bytes.NewReader(body)) + req.Header.Set("x-ms-blob-type", "BlockBlob") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + return resp.StatusCode +} + +// save runs the client's whole save flow; blocks > 1 uses Put Block in +// REVERSE order (the SDK uploads concurrently) and a Put Block List. +func (h *harness) save(key, version string, data []byte, blocks int) map[string]any { + h.t.Helper() + cr := h.twirp("CreateCacheEntry", map[string]any{"key": key, "version": version}) + if cr["ok"] != true { + return cr + } + up := cr["signed_upload_url"].(string) + if blocks <= 1 { + if code := put(h.t, up, data); code != http.StatusCreated { + h.t.Fatalf("Put Blob: %d", code) + } + } else { + size := (len(data) + blocks - 1) / blocks + var ids []string + for i := 0; i < blocks; i++ { + ids = append(ids, base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("block-%06d", i)))) + } + for i := blocks - 1; i >= 0; i-- { + lo, hi := i*size, (i+1)*size + if hi > len(data) { + hi = len(data) + } + if code := put(h.t, up+"?comp=block&blockid="+ids[i], data[lo:hi]); code != http.StatusCreated { + h.t.Fatalf("Put Block %d: %d", i, code) + } + } + var xml strings.Builder + xml.WriteString(``) + for _, id := range ids { + xml.WriteString("" + id + "") + } + xml.WriteString("") + if code := put(h.t, up+"?comp=blocklist", []byte(xml.String())); code != http.StatusCreated { + h.t.Fatalf("Put Block List: %d", code) + } + } + return h.twirp("FinalizeCacheEntryUpload", map[string]any{ + "key": key, "version": version, "size_bytes": strconv.Itoa(len(data)), + }) +} + +// restore runs the client's restore flow and returns (matched key, bytes), +// or ("", nil) on a miss. +func (h *harness) restore(key string, restoreKeys []string, version string) (string, []byte) { + h.t.Helper() + rk := []any{} + for _, k := range restoreKeys { + rk = append(rk, k) + } + r := h.twirp("GetCacheEntryDownloadURL", map[string]any{"key": key, "restore_keys": rk, "version": version}) + if r["ok"] != true { + return "", nil + } + resp, err := http.Get(r["signed_download_url"].(string)) + if err != nil { + h.t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + data, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + h.t.Fatalf("download: %s", resp.Status) + } + if cl := resp.Header.Get("Content-Length"); cl != strconv.Itoa(len(data)) { + h.t.Fatalf("Content-Length %s but %d bytes arrived; the client rejects that as incomplete", cl, len(data)) + } + return r["matched_key"].(string), data +} + +func payload(n int) []byte { + b := make([]byte, n) + for i := range b { + b[i] = byte(i*31 + i/7) + } + return b +} + +var mainScope = Scope{Repo: "ephpm/ephpm", Write: "refs/heads/main", Reads: []string{"refs/heads/main"}} + +func TestSaveThenRestore_SingleShot(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + + data := payload(3000) + if fr := h.save("cargo-linux-abc", "v1", data, 1); fr["ok"] != true { + t.Fatalf("finalize: %v", fr) + } + key, got := h.restore("cargo-linux-abc", nil, "v1") + if key != "cargo-linux-abc" || !bytes.Equal(got, data) { + t.Fatalf("restore = %q (%d bytes), want the saved entry back byte for byte", key, len(got)) + } +} + +func TestSaveThenRestore_ConcurrentBlocksOutOfOrder(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + + data := payload(10_007) // not a multiple of the block size + if fr := h.save("k", "v", data, 5); fr["ok"] != true { + t.Fatalf("finalize: %v", fr) + } + _, got := h.restore("k", nil, "v") + if !bytes.Equal(got, data) { + t.Fatal("blocks uploaded in reverse came back in the wrong order; the block list must decide order") + } + // Each block streamed through as its own blob (5 blocks + 1 config). + if h.reg.patches != 6 { + t.Errorf("registry saw %d blob uploads, want 6 (one per block plus the config)", h.reg.patches) + } +} + +func TestRestore_RangeAcrossBlocks(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + data := payload(1000) + h.save("k", "v", data, 4) + + r := h.twirp("GetCacheEntryDownloadURL", map[string]any{"key": "k", "version": "v"}) + req, _ := http.NewRequest(http.MethodGet, r["signed_download_url"].(string), nil) + req.Header.Set("Range", "bytes=240-760") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + got, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusPartialContent || !bytes.Equal(got, data[240:761]) { + t.Fatalf("range 240-760 = %s, %d bytes; want 206 and the exact slice", resp.Status, len(got)) + } + if cr := resp.Header.Get("Content-Range"); cr != "bytes 240-760/1000" { + t.Errorf("Content-Range = %q", cr) + } +} + +func TestRestoreKeys_PrefixNewestWinsAndVersionMustMatch(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + + h.save("cargo-linux-aaa", "v1", []byte("old"), 1) + time.Sleep(5 * time.Millisecond) + h.save("cargo-linux-bbb", "v1", []byte("new"), 1) + h.save("cargo-linux-ccc", "v2", []byte("other version"), 1) + + key, got := h.restore("cargo-linux-zzz", []string{"cargo-linux-"}, "v1") + if key != "cargo-linux-bbb" || string(got) != "new" { + t.Fatalf("prefix restore = %q %q; want the NEWEST v1 entry", key, got) + } + if key, _ := h.restore("cargo-linux-ccc", nil, "v1"); key != "" { + t.Fatalf("exact key with a different version matched %q; versions must match", key) + } + if key, _ := h.restore("nope", []string{"zzz-"}, "v1"); key != "" { + t.Fatalf("no match should miss, got %q", key) + } +} + +func TestScope_PRReadsMainButCannotWriteIt(t *testing.T) { + h := newHarness(t) + h.p.Track("main-runner") + h.p.Bind("main-runner", mainScope) + h.as("main-runner") + h.save("deps-1", "v", []byte("from main"), 1) + + pr := Scope{Repo: "ephpm/ephpm", Write: "refs/pull/7/merge", Reads: []string{"refs/pull/7/merge", "refs/heads/main"}} + h.p.Track("pr-runner") + h.p.Bind("pr-runner", pr) + h.as("pr-runner") + + if _, got := h.restore("deps-1", nil, "v"); string(got) != "from main" { + t.Fatalf("PR restore of main's entry = %q; a PR must read the default branch", got) + } + // The PR saves a NEW key: it lands in the PR's scope only. + if fr := h.save("deps-2", "v", []byte("from pr"), 1); fr["ok"] != true { + t.Fatalf("PR save into its own scope: %v", fr) + } + h.as("main-runner") + if key, _ := h.restore("deps-2", nil, "v"); key != "" { + t.Fatal("main restored an entry a PR saved; PR writes must never be visible to main") + } + if n := h.reg.tagCount(scopeRepository("ephpm/ephpm", "refs/heads/main")); n != 1 { + t.Fatalf("main's scope holds %d entries, want exactly main's own 1", n) + } +} + +func TestScope_ReadOnlyJobCannotSave(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", Scope{Repo: "ephpm/ephpm", Reads: []string{"refs/heads/main"}}) + if cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k", "version": "v"}); cr["ok"] != false { + t.Fatalf("read-only job reserved an entry: %v", cr) + } +} + +func TestScope_ReposAreIsolated(t *testing.T) { + h := newHarness(t) + h.p.Track("a") + h.p.Bind("a", Scope{Repo: "org/foo.bar", Write: "refs/heads/main", Reads: []string{"refs/heads/main"}}) + h.as("a") + h.save("k", "v", []byte("foo.bar"), 1) + + // "foo-bar" squashes to the same path component; the hash must keep the + // namespaces apart. + h.p.Track("b") + h.p.Bind("b", Scope{Repo: "org/foo-bar", Write: "refs/heads/main", Reads: []string{"refs/heads/main"}}) + h.as("b") + if key, _ := h.restore("k", nil, "v"); key != "" { + t.Fatal("a different repo restored another repo's entry") + } +} + +func TestUnboundOrUnknownJobGetsMissesAndNoSaves(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + h.save("k", "v", []byte("x"), 1) + + h.p.Track("fresh") // tracked, not yet bound by in_progress + h.as("fresh") + if key, _ := h.restore("k", nil, "v"); key != "" { + t.Fatal("an unbound runner got a hit") + } + if cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k2", "version": "v"}); cr["ok"] != false { + t.Fatal("an unbound runner reserved an entry") + } + h.as("stranger") // not tracked at all + if key, _ := h.restore("k", nil, "v"); key != "" { + t.Fatal("an untracked source got a hit") + } +} + +func TestSignedURLsAreBoundToTheirJob(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + h.save("k", "v", []byte("secret build"), 1) + r := h.twirp("GetCacheEntryDownloadURL", map[string]any{"key": "k", "version": "v"}) + url := r["signed_download_url"].(string) + + h.p.Track("other") + h.p.Bind("other", Scope{Repo: "evil/repo", Write: "refs/heads/main", Reads: []string{"refs/heads/main"}}) + h.as("other") + resp, err := http.Get(url) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("another job fetched a download URL it was not granted: %s", resp.Status) + } +} + +func TestConflictingRebindRevokes(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + h.p.Bind("runner-1", Scope{Repo: "ephpm/ephpm", Write: "refs/pull/9/merge", Reads: []string{"refs/pull/9/merge"}}) + if cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k", "version": "v"}); cr["ok"] != false { + t.Fatal("a runner bound twice to different scopes kept write access") + } +} + +func TestEntriesAreImmutable(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + h.save("k", "v", []byte("first"), 1) + if cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k", "version": "v"}); cr["ok"] != false { + t.Fatal("a second save of an existing key was accepted") + } +} + +func TestFinalizeRejectsSizeMismatch(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k", "version": "v"}) + _ = put(t, cr["signed_upload_url"].(string), []byte("12345")) + fr := h.twirp("FinalizeCacheEntryUpload", map[string]any{"key": "k", "version": "v", "size_bytes": 99}) + if fr["ok"] != false { + t.Fatal("finalize accepted a size that does not match what was uploaded") + } +} + +func TestEntrySizeLimit(t *testing.T) { + h := newHarness(t, func(c *Config) { c.MaxEntryBytes = 100 }) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k", "version": "v"}) + if code := put(t, cr["signed_upload_url"].(string), payload(101)); code >= 500 || code < 400 { + t.Fatalf("oversize upload answered %d; want a non-retryable 4xx", code) + } + if fr := h.twirp("FinalizeCacheEntryUpload", map[string]any{"key": "k", "version": "v", "size_bytes": 101}); fr["ok"] != false { + t.Fatal("an oversize entry was finalized") + } +} + +func TestRegistryDownIsAMiss(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + h.save("k", "v", []byte("x"), 1) + + h.reg.setDown(true) + h.p.probe(context.Background()) + if h.p.Healthy() { + t.Fatal("still healthy with the registry down") + } + if key, _ := h.restore("k", nil, "v"); key != "" { + t.Fatal("got a hit with the registry down") + } + if cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k2", "version": "v"}); cr["ok"] != false { + t.Fatal("reserved an entry with the registry down") + } +} + +func TestEvictionDropsLeastRecentlyUsed(t *testing.T) { + h := newHarness(t, func(c *Config) { c.MaxBytesPerRepo = 250 }) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + + h.save("a", "v", payload(100), 1) + time.Sleep(5 * time.Millisecond) + h.save("b", "v", payload(100), 1) + time.Sleep(5 * time.Millisecond) + h.restore("a", nil, "v") // a is now the most recently USED + time.Sleep(5 * time.Millisecond) + h.save("c", "v", payload(100), 1) // 300 > 250: one must go + + if key, _ := h.restore("b", nil, "v"); key != "" { + t.Fatal("b (least recently used) survived eviction") + } + for _, k := range []string{"a", "c"} { + if key, _ := h.restore(k, nil, "v"); key != k { + t.Fatalf("%s was evicted; only the least recently used entry should go", k) + } + } +} + +func TestForgetRevokesEverything(t *testing.T) { + h := newHarness(t) + h.p.Track("runner-1") + h.p.Bind("runner-1", mainScope) + cr := h.twirp("CreateCacheEntry", map[string]any{"key": "k", "version": "v"}) + h.p.Forget("runner-1") + if code := put(t, cr["signed_upload_url"].(string), []byte("x")); code != http.StatusNotFound { + t.Fatalf("upload after Forget answered %d; want 404", code) + } +} + +func TestRelayPassesNonCacheCallsToGitHub(t *testing.T) { + var gotAuth, gotPath, gotHost string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth, gotPath, gotHost = r.Header.Get("Authorization"), r.URL.Path, r.Host + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer upstream.Close() + h := newHarness(t, func(c *Config) { c.UpstreamResultsURL = upstream.URL }) + h.p.Track("runner-1") + + const art = "/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact" + req, _ := http.NewRequest(http.MethodPost, h.srv.URL+art, strings.NewReader(`{}`)) + req.Header.Set("Authorization", "Bearer job-token") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK || gotPath != art || gotAuth != "Bearer job-token" { + t.Fatalf("relay: status %s path %q auth %q; artifacts must reach GitHub unchanged", resp.Status, gotPath, gotAuth) + } + if !strings.Contains(upstream.URL, gotHost) { + t.Errorf("relay sent Host %q, want the upstream's", gotHost) + } + + h.as("stranger") + resp, err = http.Post(h.srv.URL+art, "application/json", strings.NewReader(`{}`)) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("an untracked source used the relay: %s; it must not be an open relay", resp.Status) + } +} + +func TestTwirpAcceptsNumericAndStringInt64(t *testing.T) { + for _, raw := range []string{`{"size_bytes":"12"}`, `{"size_bytes":12}`} { + var r twirpReq + if err := json.Unmarshal([]byte(raw), &r); err != nil || r.SizeBytes != 12 { + t.Errorf("%s -> %d, %v", raw, r.SizeBytes, err) + } + } +} + +func TestParseBlockList(t *testing.T) { + ids, err := parseBlockList(strings.NewReader(`QQ==Qg==Qw==`)) + if err != nil || strings.Join(ids, ",") != "QQ==,Qg==,Qw==" { + t.Fatalf("ids = %v, %v", ids, err) + } + for _, bad := range []string{``, ``, `not base64!`, `QQ==`} { + if _, err := parseBlockList(strings.NewReader(bad)); err == nil { + t.Errorf("parseBlockList(%q) accepted a bad block list", bad) + } + } +} + +func TestParseRange(t *testing.T) { + for _, tc := range []struct { + h string + s, e int64 + ok bool + }{ + {"bytes=0-9", 0, 9, true}, + {"bytes=5-", 5, 99, true}, + {"bytes=-10", 90, 99, true}, + {"bytes=90-200", 90, 99, true}, + {"bytes=100-", 0, 0, false}, + {"bytes=5-2", 0, 0, false}, + {"bytes=0-1,4-5", 0, 0, false}, + {"items=0-1", 0, 0, false}, + } { + s, e, ok := parseRange(tc.h, 100) + if ok != tc.ok || (ok && (s != tc.s || e != tc.e)) { + t.Errorf("parseRange(%q) = %d,%d,%v want %d,%d,%v", tc.h, s, e, ok, tc.s, tc.e, tc.ok) + } + } +} + +func TestNaming(t *testing.T) { + ns := repoNamespace("EPHPM/ephpm") + if ns != repoNamespace("ephpm/ephpm") { + t.Error("repo namespace is case-sensitive; GitHub repo names are not") + } + if repoNamespace("org/foo.bar") == repoNamespace("org/foo-bar") { + t.Error("two distinct repos share a namespace") + } + if scopeRepository("o/r", "refs/heads/main") == scopeRepository("o/r", "refs/pull/1/merge") { + t.Error("two refs share a scope repository") + } + if entryTag("k", "v1") == entryTag("k", "v2") { + t.Error("version is not part of the tag") + } + if tag := entryTag(strings.Repeat("x", 512), "v"); len(tag) > 128 { + t.Errorf("tag %d chars; registry tags are at most 128", len(tag)) + } + for _, s := range []string{ns, scopeRepository("Weird__Org/..repo..", "refs/heads/a/b")} { + for _, comp := range strings.Split(s, "/") { + if comp == "" || strings.Trim(comp, "abcdefghijklmnopqrstuvwxyz0123456789-") != "" || strings.HasPrefix(comp, "-") || strings.HasSuffix(comp, "-") || strings.Contains(comp, "--") { + t.Errorf("%q has an invalid registry path component %q", s, comp) + } + } + } +} + +// TestStart_BeforeTheBridgeExists pins the fresh-boot case found on a live +// run: the daemon starts this server before any job has brought the CNI +// bridge (and so the gateway IP) up, and a plain bind failed with +// "cannot assign requested address", leaving every job on GitHub's cache. +// 192.0.2.1 (TEST-NET-1) is never assigned locally, so Start must fall back +// to the wildcard on the same port rather than fail. +func TestStart_BeforeTheBridgeExists(t *testing.T) { + reg, _ := newFakeRegistry(t) + regSrv := httptest.NewServer(reg) + t.Cleanup(regSrv.Close) + p, err := New(Config{ + ListenAddr: "192.0.2.1:0", + AdvertiseBase: "http://192.0.2.1:8089", + Registry: regSrv.URL, + RegistryUser: "cache", + RegistryPassword: "s3cret", + HealthInterval: -1, + JobForIP: func(string) string { return "" }, + }) + if err != nil { + t.Fatal(err) + } + if err := p.Start(); err != nil { + t.Fatalf("Start with the gateway address not yet assigned: %v", err) + } + t.Cleanup(func() { _ = p.Stop() }) + if got := p.ResultsURL(); got != "http://192.0.2.1:8089/" { + t.Errorf("ResultsURL = %q; it must stay the advertised gateway, not the wildcard bind", got) + } +} diff --git a/pkg/proxies/actionscache/fakeregistry_test.go b/pkg/proxies/actionscache/fakeregistry_test.go new file mode 100644 index 0000000..c2ff28f --- /dev/null +++ b/pkg/proxies/actionscache/fakeregistry_test.go @@ -0,0 +1,219 @@ +package actionscache + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "sort" + "strconv" + "strings" + "sync" + "testing" +) + +// fakeRegistry is the subset of registry:2 the cache uses, in memory. Like +// the real one it refuses a manifest whose config or layers are missing, so +// a test cannot pass by writing manifests that point at nothing. +type fakeRegistry struct { + t *testing.T + user string + pass string + mu sync.Mutex + blobs map[string][]byte // digest -> bytes (global, like a shared store) + repoBlob map[string]map[string]bool // repo -> digests linked into it + uploads map[string]*strings.Builder // upload id -> data + uploadTo map[string]string // upload id -> repo + tags map[string]map[string]string // repo -> tag -> manifest digest + mans map[string][]byte // manifest digest -> body + nextID int + down bool + patches int +} + +func newFakeRegistry(t *testing.T) (*fakeRegistry, *httptest.Server) { + f := &fakeRegistry{ + t: t, user: "cache", pass: "s3cret", + blobs: map[string][]byte{}, repoBlob: map[string]map[string]bool{}, + uploads: map[string]*strings.Builder{}, uploadTo: map[string]string{}, + tags: map[string]map[string]string{}, mans: map[string][]byte{}, + } + srv := httptest.NewServer(f) + t.Cleanup(srv.Close) + return f, srv +} + +func digestOf(b []byte) string { + s := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(s[:]) +} + +func (f *fakeRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + if f.down { + http.Error(w, "down", http.StatusServiceUnavailable) + return + } + if u, p, ok := r.BasicAuth(); !ok || u != f.user || p != f.pass { + w.Header().Set("WWW-Authenticate", `Basic realm="registry"`) + http.Error(w, "auth", http.StatusUnauthorized) + return + } + path := r.URL.Path + switch { + case path == "/v2/": + w.WriteHeader(http.StatusOK) + case path == "/v2/_catalog": + var names []string + for repo := range f.tags { + names = append(names, repo) + } + sort.Strings(names) + _ = json.NewEncoder(w).Encode(map[string]any{"repositories": names}) + case strings.HasSuffix(path, "/blobs/uploads/") && r.Method == http.MethodPost: + repo := strings.TrimSuffix(strings.TrimPrefix(path, "/v2/"), "/blobs/uploads/") + f.nextID++ + id := strconv.Itoa(f.nextID) + f.uploads[id] = &strings.Builder{} + f.uploadTo[id] = repo + w.Header().Set("Location", "/v2/"+repo+"/blobs/uploads/"+id+"?_state=x") + w.WriteHeader(http.StatusAccepted) + case strings.Contains(path, "/blobs/uploads/") && r.Method == http.MethodPatch: + id := path[strings.LastIndex(path, "/")+1:] + b, ok := f.uploads[id] + if !ok { + http.NotFound(w, r) + return + } + f.patches++ + data, _ := io.ReadAll(r.Body) + b.Write(data) + w.Header().Set("Location", path+"?_state=y") + w.WriteHeader(http.StatusAccepted) + case strings.Contains(path, "/blobs/uploads/") && r.Method == http.MethodPut: + id := path[strings.LastIndex(path, "/")+1:] + b, ok := f.uploads[id] + if !ok { + http.NotFound(w, r) + return + } + data := []byte(b.String()) + want := r.URL.Query().Get("digest") + if got := digestOf(data); got != want { + http.Error(w, "digest mismatch: got "+got+" want "+want, http.StatusBadRequest) + return + } + repo := f.uploadTo[id] + f.blobs[want] = data + if f.repoBlob[repo] == nil { + f.repoBlob[repo] = map[string]bool{} + } + f.repoBlob[repo][want] = true + delete(f.uploads, id) + w.WriteHeader(http.StatusCreated) + case strings.Contains(path, "/blobs/") && r.Method == http.MethodGet: + repo, dig, _ := strings.Cut(strings.TrimPrefix(path, "/v2/"), "/blobs/") + if !f.repoBlob[repo][dig] { + http.NotFound(w, r) + return + } + data := f.blobs[dig] + if rh := r.Header.Get("Range"); rh != "" { + s, e, ok := parseRange(rh, int64(len(data))) + if !ok { + http.Error(w, "range", http.StatusRequestedRangeNotSatisfiable) + return + } + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write(data[s : e+1]) + return + } + _, _ = w.Write(data) + case strings.HasSuffix(path, "/tags/list"): + repo := strings.TrimSuffix(strings.TrimPrefix(path, "/v2/"), "/tags/list") + tags, ok := f.tags[repo] + if !ok { + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"errors":[{"code":"NAME_UNKNOWN"}]}`)) + return + } + var names []string + for t := range tags { + names = append(names, t) + } + sort.Strings(names) + _ = json.NewEncoder(w).Encode(map[string]any{"name": repo, "tags": names}) + case strings.Contains(path, "/manifests/"): + repo, ref, _ := strings.Cut(strings.TrimPrefix(path, "/v2/"), "/manifests/") + switch r.Method { + case http.MethodPut: + body, _ := io.ReadAll(r.Body) + var m manifest + if err := json.Unmarshal(body, &m); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + for _, d := range append([]descriptor{m.Config}, m.Layers...) { + if !f.repoBlob[repo][d.Digest] { + http.Error(w, fmt.Sprintf(`{"errors":[{"code":"MANIFEST_BLOB_UNKNOWN","detail":%q}]}`, d.Digest), http.StatusBadRequest) + return + } + } + dig := digestOf(body) + f.mans[dig] = body + if f.tags[repo] == nil { + f.tags[repo] = map[string]string{} + } + f.tags[repo][ref] = dig + w.Header().Set("Docker-Content-Digest", dig) + w.WriteHeader(http.StatusCreated) + case http.MethodGet: + dig := ref + if !strings.HasPrefix(ref, "sha256:") { + dig = f.tags[repo][ref] + } + body, ok := f.mans[dig] + if !ok { + http.NotFound(w, r) + return + } + w.Header().Set("Docker-Content-Digest", dig) + _, _ = w.Write(body) + case http.MethodDelete: + found := false + for t, d := range f.tags[repo] { + if d == ref { + delete(f.tags[repo], t) + found = true + } + } + if len(f.tags[repo]) == 0 { + delete(f.tags, repo) + } + if !found { + http.NotFound(w, r) + return + } + w.WriteHeader(http.StatusAccepted) + } + default: + f.t.Errorf("fake registry: unexpected %s %s", r.Method, r.URL) + http.NotFound(w, r) + } +} + +func (f *fakeRegistry) setDown(down bool) { + f.mu.Lock() + f.down = down + f.mu.Unlock() +} + +func (f *fakeRegistry) tagCount(repo string) int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.tags[repo]) +} diff --git a/pkg/proxies/actionscache/index.go b/pkg/proxies/actionscache/index.go new file mode 100644 index 0000000..adb5d54 --- /dev/null +++ b/pkg/proxies/actionscache/index.go @@ -0,0 +1,343 @@ +package actionscache + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +// Manifest annotations carrying an entry's identity. The tag is only a hash, +// so these are the source of truth for key/version/ref. +const ( + annKey = "dev.ephemerd.actions-cache.key" + annVersion = "dev.ephemerd.actions-cache.version" + annRef = "dev.ephemerd.actions-cache.ref" + annCreated = "dev.ephemerd.actions-cache.created" + annSize = "dev.ephemerd.actions-cache.size" +) + +type descriptor struct { + MediaType string `json:"mediaType"` + Digest string `json:"digest"` + Size int64 `json:"size"` +} + +type manifest struct { + SchemaVersion int `json:"schemaVersion"` + MediaType string `json:"mediaType"` + Config descriptor `json:"config"` + Layers []descriptor `json:"layers"` + Annotations map[string]string `json:"annotations"` +} + +// entry is one saved cache archive. Its bytes are Layers concatenated in +// order — one layer per uploaded block, so concurrent block uploads never +// have to be reassembled before they reach the registry. +type entry struct { + Key string + Version string + Ref string + Created time.Time + Size int64 + Layers []descriptor + + repository string // registry repository (scopeRepository of Ref) + tag string + digest string // manifest digest, for deletion + + // lastUsed is when this node last served or saved the entry. In memory + // only, so after a restart (or for entries another node wrote) eviction + // falls back to Created. + lastUsed time.Time +} + +func (e *entry) recency() time.Time { + if e.lastUsed.After(e.Created) { + return e.lastUsed + } + return e.Created +} + +func encodeManifest(e *entry, cfg descriptor) ([]byte, error) { + m := manifest{ + SchemaVersion: 2, + MediaType: manifestMediaType, + Config: cfg, + Layers: e.Layers, + Annotations: map[string]string{ + annKey: e.Key, + annVersion: e.Version, + annRef: e.Ref, + annCreated: e.Created.UTC().Format(time.RFC3339Nano), + annSize: strconv.FormatInt(e.Size, 10), + }, + } + if m.Layers == nil { + m.Layers = []descriptor{} + } + return json.Marshal(m) +} + +func decodeManifest(body []byte) (*entry, error) { + var m manifest + if err := json.Unmarshal(body, &m); err != nil { + return nil, err + } + if m.Config.MediaType != configMediaType { + return nil, fmt.Errorf("not an actions-cache manifest (config %q)", m.Config.MediaType) + } + e := &entry{ + Key: m.Annotations[annKey], + Version: m.Annotations[annVersion], + Ref: m.Annotations[annRef], + Layers: m.Layers, + } + e.Created, _ = time.Parse(time.RFC3339Nano, m.Annotations[annCreated]) + for _, l := range m.Layers { + e.Size += l.Size + } + if want, err := strconv.ParseInt(m.Annotations[annSize], 10, 64); err == nil && want != e.Size { + return nil, fmt.Errorf("layers sum to %d bytes, annotation says %d", e.Size, want) + } + return e, nil +} + +// configBlob is the small config object every manifest points at. registry:2 +// insists the config blob exists; this one carries the same identity as the +// annotations, which also makes every entry's config digest distinct. +func configBlob(e *entry) []byte { + b, _ := json.Marshal(map[string]string{ + "key": e.Key, + "version": e.Version, + "ref": e.Ref, + "created": e.Created.UTC().Format(time.RFC3339Nano), + }) + return b +} + +// scopeIndex is the in-memory view of one registry repository (one ref of +// one forge repo): tag -> entry. It is refreshed from the registry's tag list +// on every lookup, because other nodes write to the same registry; only tags +// it has not seen cost a manifest fetch. +type scopeIndex struct { + mu sync.Mutex + entries map[string]*entry +} + +// index is every scopeIndex this node has touched, plus the bookkeeping for +// per-repo eviction. +type index struct { + reg *registry + + mu sync.Mutex + scopes map[string]*scopeIndex // registry repository -> index + // lastSweep is when a repo namespace's full catalog was last walked for + // eviction; the walk is rate-limited because the catalog is global. + lastSweep map[string]time.Time +} + +func newIndex(reg *registry) *index { + return &index{reg: reg, scopes: map[string]*scopeIndex{}, lastSweep: map[string]time.Time{}} +} + +func (ix *index) scope(repository string) *scopeIndex { + ix.mu.Lock() + defer ix.mu.Unlock() + s, ok := ix.scopes[repository] + if !ok { + s = &scopeIndex{entries: map[string]*entry{}} + ix.scopes[repository] = s + } + return s +} + +// refresh syncs one repository's index with the registry and returns a +// snapshot of its entries. +func (ix *index) refresh(ctx context.Context, repository string) ([]*entry, error) { + tags, err := ix.reg.listTags(ctx, repository) + if err != nil { + return nil, err + } + s := ix.scope(repository) + s.mu.Lock() + known := make(map[string]*entry, len(s.entries)) + for t, e := range s.entries { + known[t] = e + } + s.mu.Unlock() + + live := make(map[string]*entry, len(tags)) + for _, t := range tags { + if e, ok := known[t]; ok { + live[t] = e + continue + } + body, digest, err := ix.reg.getManifest(ctx, repository, t) + if err != nil { + continue // deleted between list and fetch, or not ours + } + e, err := decodeManifest(body) + if err != nil { + continue + } + e.repository, e.tag, e.digest = repository, t, digest + live[t] = e + } + + s.mu.Lock() + s.entries = live + out := make([]*entry, 0, len(live)) + for _, e := range live { + out = append(out, e) + } + s.mu.Unlock() + return out, nil +} + +func (ix *index) add(e *entry) { + s := ix.scope(e.repository) + s.mu.Lock() + s.entries[e.tag] = e + s.mu.Unlock() +} + +func (ix *index) remove(e *entry) { + s := ix.scope(e.repository) + s.mu.Lock() + if cur, ok := s.entries[e.tag]; ok && cur.digest == e.digest { + delete(s.entries, e.tag) + } + s.mu.Unlock() +} + +// lookup finds the entry a restore should get, with GitHub's matching rules: +// for each readable ref in order, an exact key match first, then each restore +// key as a prefix in order, newest entry winning a prefix match. An entry +// only ever matches its own version — the same key with other paths or +// compression is a different cache. +func (ix *index) lookup(ctx context.Context, s Scope, key string, restoreKeys []string, version string) (*entry, error) { + var firstErr error + for _, ref := range s.Reads { + entries, err := ix.refresh(ctx, scopeRepository(s.Repo, ref)) + if err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + if e := match(entries, key, restoreKeys, version); e != nil { + return e, nil + } + } + return nil, firstErr +} + +func match(entries []*entry, key string, restoreKeys []string, version string) *entry { + for _, e := range entries { + if e.Version == version && e.Key == key { + return e + } + } + for _, rk := range restoreKeys { + if rk == "" { + continue + } + var best *entry + for _, e := range entries { + if e.Version != version || !strings.HasPrefix(e.Key, rk) { + continue + } + if best == nil || e.Created.After(best.Created) { + best = e + } + } + if best != nil { + return best + } + } + return nil +} + +// exists reports whether (key, version) is already saved under ref. Entries +// are immutable, as on GitHub: a second save of the same key is refused. +func (ix *index) exists(ctx context.Context, repo, ref, key, version string) (bool, error) { + entries, err := ix.refresh(ctx, scopeRepository(repo, ref)) + if err != nil { + return false, err + } + for _, e := range entries { + if e.Key == key && e.Version == version { + return true, nil + } + } + return false, nil +} + +// evict deletes the least recently used entries of repo until its total size +// is within maxBytes. The full namespace walk (every ref the repo ever saved +// under) is rate-limited to once per sweepEvery; between walks only the +// scopes this node has touched are counted, which can only under-count and +// so only ever delays an eviction — it never deletes too much. +func (ix *index) evict(ctx context.Context, repo string, maxBytes int64, sweepEvery time.Duration, now time.Time) ([]*entry, error) { + if maxBytes <= 0 { + return nil, nil + } + ns := repoNamespace(repo) + "/" + + ix.mu.Lock() + walk := now.Sub(ix.lastSweep[ns]) >= sweepEvery + if walk { + ix.lastSweep[ns] = now + } + var repos []string + for r := range ix.scopes { + if strings.HasPrefix(r, ns) { + repos = append(repos, r) + } + } + ix.mu.Unlock() + + if walk { + all, err := ix.reg.catalog(ctx, ns) + if err != nil { + return nil, err + } + repos = all + } + + var entries []*entry + var total int64 + for _, r := range repos { + es, err := ix.refresh(ctx, r) + if err != nil { + continue + } + for _, e := range es { + entries = append(entries, e) + total += e.Size + } + } + if total <= maxBytes { + return nil, nil + } + + sort.Slice(entries, func(i, j int) bool { return entries[i].recency().Before(entries[j].recency()) }) + var evicted []*entry + for _, e := range entries { + if total <= maxBytes { + break + } + if err := ix.reg.deleteManifest(ctx, e.repository, e.digest); err != nil { + return evicted, err + } + ix.remove(e) + total -= e.Size + evicted = append(evicted, e) + } + return evicted, nil +} diff --git a/pkg/proxies/actionscache/oci.go b/pkg/proxies/actionscache/oci.go new file mode 100644 index 0000000..b1e9db5 --- /dev/null +++ b/pkg/proxies/actionscache/oci.go @@ -0,0 +1,368 @@ +package actionscache + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "hash" + "io" + "net/http" + "net/url" + "strconv" + "strings" +) + +// registry is the slice of the OCI distribution API the cache needs, spoken +// to a writable registry:2. Hand-rolled rather than go-containerregistry's +// remote package because entries are streamed in as blocks whose digest is +// not known until the last byte, and the whole surface is seven endpoints. +type registry struct { + base *url.URL + user string + pass string + hc *http.Client +} + +const ( + manifestMediaType = "application/vnd.oci.image.manifest.v1+json" + configMediaType = "application/vnd.ephemerd.actions-cache.config.v1+json" + chunkMediaType = "application/vnd.ephemerd.actions-cache.chunk.v1" +) + +// errNotFound is a 404 from the registry (unknown repository, tag or blob). +var errNotFound = errors.New("not found in registry") + +func newRegistry(endpoint, user, pass string, hc *http.Client) (*registry, error) { + u, err := url.Parse(strings.TrimRight(endpoint, "/")) + if err != nil || u.Scheme == "" || u.Host == "" { + return nil, fmt.Errorf("registry endpoint %q is not an absolute URL", endpoint) + } + if hc == nil { + hc = http.DefaultClient + } + return ®istry{base: u, user: user, pass: pass, hc: hc}, nil +} + +func (r *registry) url(path string) string { + return r.base.String() + path +} + +func (r *registry) do(req *http.Request) (*http.Response, error) { + if r.user != "" { + req.SetBasicAuth(r.user, r.pass) + } + return r.hc.Do(req) +} + +// status turns a non-2xx response into an error, consuming the body. +func status(resp *http.Response, what string) error { + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + if resp.StatusCode == http.StatusNotFound { + return fmt.Errorf("%s: %w", what, errNotFound) + } + return fmt.Errorf("%s: registry answered %s: %s", what, resp.Status, strings.TrimSpace(string(body))) +} + +func drain(resp *http.Response) { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16)) + _ = resp.Body.Close() +} + +// ping is the health probe: the registry's API root, authenticated. +func (r *registry) ping(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.url("/v2/"), nil) + if err != nil { + return err + } + resp, err := r.do(req) + if err != nil { + return err + } + defer drain(resp) + return status(resp, "GET /v2/") +} + +// resolve turns an upload Location (relative or absolute) into a URL. +func (r *registry) resolve(loc string) (*url.URL, error) { + u, err := url.Parse(loc) + if err != nil { + return nil, fmt.Errorf("bad upload location %q: %w", loc, err) + } + return r.base.ResolveReference(u), nil +} + +// countingHasher digests and counts what is streamed through it. +type countingHasher struct { + h hash.Hash + n int64 +} + +func (c *countingHasher) Write(p []byte) (int, error) { + c.n += int64(len(p)) + return c.h.Write(p) +} + +// uploadBlob streams body into the repository as one blob: start an upload, +// PATCH the bytes through (digesting them on the way), then close it with the +// digest. It never holds more than the transport's buffer in memory. +func (r *registry) uploadBlob(ctx context.Context, repo string, body io.Reader) (digest string, size int64, err error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.url("/v2/"+repo+"/blobs/uploads/"), nil) + if err != nil { + return "", 0, err + } + resp, err := r.do(req) + if err != nil { + return "", 0, err + } + drain(resp) + if resp.StatusCode != http.StatusAccepted { + return "", 0, fmt.Errorf("starting blob upload in %s: registry answered %s", repo, resp.Status) + } + loc, err := r.resolve(resp.Header.Get("Location")) + if err != nil { + return "", 0, err + } + + ch := &countingHasher{h: sha256.New()} + req, err = http.NewRequestWithContext(ctx, http.MethodPatch, loc.String(), io.TeeReader(body, ch)) + if err != nil { + return "", 0, err + } + req.Header.Set("Content-Type", "application/octet-stream") + resp, err = r.do(req) + if err != nil { + return "", 0, err + } + drain(resp) + if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusNoContent { + return "", 0, fmt.Errorf("streaming blob into %s: registry answered %s", repo, resp.Status) + } + if l := resp.Header.Get("Location"); l != "" { + if loc, err = r.resolve(l); err != nil { + return "", 0, err + } + } + + digest = "sha256:" + hex.EncodeToString(ch.h.Sum(nil)) + q := loc.Query() + q.Set("digest", digest) + loc.RawQuery = q.Encode() + req, err = http.NewRequestWithContext(ctx, http.MethodPut, loc.String(), http.NoBody) + if err != nil { + return "", 0, err + } + resp, err = r.do(req) + if err != nil { + return "", 0, err + } + defer drain(resp) + if resp.StatusCode != http.StatusCreated { + return "", 0, status(resp, "committing blob in "+repo) + } + return digest, ch.n, nil +} + +// putManifest stores body under tag and returns the digest the registry +// filed it under. +func (r *registry) putManifest(ctx context.Context, repo, tag string, body []byte) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPut, r.url("/v2/"+repo+"/manifests/"+tag), bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", manifestMediaType) + resp, err := r.do(req) + if err != nil { + return "", err + } + defer drain(resp) + if err := status(resp, "PUT manifest "+repo+":"+tag); err != nil { + return "", err + } + if d := resp.Header.Get("Docker-Content-Digest"); d != "" { + return d, nil + } + sum := sha256.Sum256(body) + return "sha256:" + hex.EncodeToString(sum[:]), nil +} + +// getManifest fetches a manifest by tag or digest. +func (r *registry) getManifest(ctx context.Context, repo, ref string) (body []byte, digest string, err error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.url("/v2/"+repo+"/manifests/"+ref), nil) + if err != nil { + return nil, "", err + } + req.Header.Set("Accept", manifestMediaType) + resp, err := r.do(req) + if err != nil { + return nil, "", err + } + defer drain(resp) + if err := status(resp, "GET manifest "+repo+":"+ref); err != nil { + return nil, "", err + } + body, err = io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if err != nil { + return nil, "", err + } + digest = resp.Header.Get("Docker-Content-Digest") + if digest == "" { + sum := sha256.Sum256(body) + digest = "sha256:" + hex.EncodeToString(sum[:]) + } + return body, digest, nil +} + +// deleteManifest removes a manifest by digest. The registry needs +// REGISTRY_STORAGE_DELETE_ENABLED=true; blobs are reclaimed later by its +// garbage collector (run from mayfly's timer, not from here). +func (r *registry) deleteManifest(ctx context.Context, repo, digest string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, r.url("/v2/"+repo+"/manifests/"+digest), nil) + if err != nil { + return err + } + resp, err := r.do(req) + if err != nil { + return err + } + defer drain(resp) + if err := status(resp, "DELETE manifest "+repo+"@"+digest); err != nil && !errors.Is(err, errNotFound) { + return err + } + return nil +} + +// listTags returns every tag in repo; an unknown repository is empty, not an +// error (no job has saved under that ref yet). +func (r *registry) listTags(ctx context.Context, repo string) ([]string, error) { + var tags []string + next := "/v2/" + repo + "/tags/list?n=1000" + for next != "" { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.url(next), nil) + if err != nil { + return nil, err + } + resp, err := r.do(req) + if err != nil { + return nil, err + } + if err := status(resp, "GET tags "+repo); err != nil { + drain(resp) + if errors.Is(err, errNotFound) { + return nil, nil + } + return nil, err + } + var page struct { + Tags []string `json:"tags"` + } + err = json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(&page) + drain(resp) + if err != nil { + return nil, fmt.Errorf("decoding tags of %s: %w", repo, err) + } + tags = append(tags, page.Tags...) + next = nextLink(resp.Header.Get("Link")) + } + return tags, nil +} + +// catalog returns the repositories whose name starts with prefix. +func (r *registry) catalog(ctx context.Context, prefix string) ([]string, error) { + var out []string + next := "/v2/_catalog?n=1000" + for next != "" { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.url(next), nil) + if err != nil { + return nil, err + } + resp, err := r.do(req) + if err != nil { + return nil, err + } + if err := status(resp, "GET catalog"); err != nil { + drain(resp) + return nil, err + } + var page struct { + Repositories []string `json:"repositories"` + } + err = json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(&page) + drain(resp) + if err != nil { + return nil, fmt.Errorf("decoding catalog: %w", err) + } + for _, name := range page.Repositories { + if strings.HasPrefix(name, prefix) { + out = append(out, name) + } + } + next = nextLink(resp.Header.Get("Link")) + } + return out, nil +} + +// nextLink extracts the path of an RFC 5988 rel="next" link, the way +// registry:2 paginates tags and the catalog. +func nextLink(h string) string { + for _, part := range strings.Split(h, ",") { + part = strings.TrimSpace(part) + if !strings.Contains(part, `rel="next"`) { + continue + } + start, end := strings.Index(part, "<"), strings.Index(part, ">") + if start < 0 || end <= start { + return "" + } + link := part[start+1 : end] + if u, err := url.Parse(link); err == nil && u.IsAbs() { + link = u.RequestURI() + } + return link + } + return "" +} + +// openBlob streams n bytes of a blob starting at off (n < 0: to the end). +func (r *registry) openBlob(ctx context.Context, repo, digest string, off, n int64) (io.ReadCloser, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.url("/v2/"+repo+"/blobs/"+digest), nil) + if err != nil { + return nil, err + } + if off > 0 || n >= 0 { + rng := "bytes=" + strconv.FormatInt(off, 10) + "-" + if n >= 0 { + rng += strconv.FormatInt(off+n-1, 10) + } + req.Header.Set("Range", rng) + } + resp, err := r.do(req) + if err != nil { + return nil, err + } + if err := status(resp, "GET blob "+repo+"@"+digest); err != nil { + drain(resp) + return nil, err + } + if (off > 0 || n >= 0) && resp.StatusCode != http.StatusPartialContent { + // A registry that ignores Range sends the whole blob: skip to off and + // bound it ourselves rather than serve the wrong bytes. + if _, err := io.CopyN(io.Discard, resp.Body, off); err != nil { + drain(resp) + return nil, fmt.Errorf("skipping to offset %d of %s: %w", off, digest, err) + } + if n >= 0 { + return struct { + io.Reader + io.Closer + }{io.LimitReader(resp.Body, n), resp.Body}, nil + } + } + return resp.Body, nil +} diff --git a/pkg/proxies/actionscache/scope.go b/pkg/proxies/actionscache/scope.go new file mode 100644 index 0000000..7dba16d --- /dev/null +++ b/pkg/proxies/actionscache/scope.go @@ -0,0 +1,82 @@ +package actionscache + +import ( + "crypto/sha256" + "encoding/hex" + "strings" +) + +// Scope is what one job may do with the cache. It is decided by ephemerd +// from the forge's own record of the job (see the scheduler's in_progress +// binding), never from anything the job sends. +type Scope struct { + // Repo is the forge-native "owner/repo" the job belongs to. Entries of + // different repos never meet: each repo gets its own registry namespace. + Repo string + // Write is the ref this job saves entries under, e.g. "refs/heads/main" + // or "refs/pull/12/merge". Empty means read-only. + Write string + // Reads are the refs a restore searches, in order — the job's own ref + // first, then the default branch. This mirrors GitHub: a PR can restore + // what main saved, but nothing a PR saves is ever visible to main. + Reads []string +} + +// repoNamespace is the registry repository prefix one forge repo's entries +// live under: actions-cache//-. +// +// Registry path components are a narrower alphabet than repo names (no +// uppercase, no "."/"_" next to another separator), so both parts are +// squashed to [a-z0-9-]. Squashing alone could map two repos onto one +// namespace ("foo.bar" and "foo-bar") and let one read the other's entries; +// the hash of the exact, case-folded name keeps them apart. +func repoNamespace(repo string) string { + repo = strings.ToLower(strings.TrimSpace(repo)) + owner, name, ok := strings.Cut(repo, "/") + if !ok { + owner, name = "_", repo + } + sum := sha256.Sum256([]byte(repo)) + return "actions-cache/" + pathComponent(owner) + "/" + pathComponent(name) + "-" + hex.EncodeToString(sum[:4]) +} + +// scopeRepository is the registry repository holding the entries one ref +// saved. The ref itself (slashes and all) is hashed into a valid component +// and kept verbatim in each entry's annotations. +func scopeRepository(repo, ref string) string { + sum := sha256.Sum256([]byte(ref)) + return repoNamespace(repo) + "/s" + hex.EncodeToString(sum[:8]) +} + +// entryTag is the tag one (key, version) pair is stored under. Cache keys +// are up to 512 arbitrary characters and tags at most 128 from a small +// alphabet, so the tag is a digest; the real key travels in the manifest. +// Version is part of it because the same key saved with different paths or +// compression is a different entry (that is what version encodes). +func entryTag(key, version string) string { + sum := sha256.Sum256([]byte(version + "\x00" + key)) + return "k" + hex.EncodeToString(sum[:20]) +} + +// pathComponent squashes s into a valid registry path component: lowercase +// alphanumerics joined by single hyphens. +func pathComponent(s string) string { + var b strings.Builder + sep := false + for _, r := range s { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9': + if sep && b.Len() > 0 { + b.WriteByte('-') + } + sep = false + b.WriteRune(r) + default: + sep = true + } + } + if b.Len() == 0 { + return "x" + } + return b.String() +} diff --git a/pkg/runner/resultsurl.go b/pkg/runner/resultsurl.go new file mode 100644 index 0000000..c32ef6d --- /dev/null +++ b/pkg/runner/resultsurl.go @@ -0,0 +1,118 @@ +package runner + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + "unicode/utf16" +) + +// The GitHub runner exports the cache/artifact service address to every +// step as ACTIONS_RESULTS_URL, taken from the job message. It sets it on the +// step's own environment, so it overwrites any value the container was +// started with — which is why ephemerd cannot simply point jobs at its LAN +// cache with an env var. +// +// PatchResultsURL renames that one string inside Runner.Worker.dll to a name +// of the same length. The runner then exports GitHub's address under the +// unused name, and the ACTIONS_RESULTS_URL ephemerd put in the container's +// environment is what the step sees. This is the technique +// falcondev-oss/github-actions-cache-server documents; the rename has to be +// same-length because the string lives in the assembly's #US heap, whose +// entries are length-prefixed. +// +// Verified against runner 2.337.0 (linux-x64 tarball and the win-x64 zip +// embedded here): Runner.Worker.dll holds exactly ONE UTF-16LE occurrence +// and Runner.Common.dll/Runner.Sdk.dll hold none. +const ( + // ResultsURLVar is the variable the unpatched runner exports. + ResultsURLVar = "ACTIONS_RESULTS_URL" + // ResultsURLRenamed is what a patched runner exports GitHub's address + // as instead. Same length as ResultsURLVar — see above. + ResultsURLRenamed = "ACTIONS_RESULTS_ORL" +) + +// WorkerDLL is the runner assembly that holds the string, relative to the +// runner root. +var WorkerDLL = filepath.Join("bin", "Runner.Worker.dll") + +var ( + resultsURLUTF16 = utf16LE(ResultsURLVar) + resultsRenamedUTF16 = utf16LE(ResultsURLRenamed) +) + +func utf16LE(s string) []byte { + units := utf16.Encode([]rune(s)) + b := make([]byte, 0, len(units)*2) + for _, u := range units { + b = append(b, byte(u), byte(u>>8)) + } + return b +} + +// PatchResultsURL rewrites every UTF-16LE occurrence of ResultsURLVar in data +// in place. It reports whether data now exports the renamed variable: true +// when it patched at least one occurrence or data was already patched, false +// when neither name is present (a runner version that no longer reads the +// variable from this assembly — the caller must then leave the cache off). +func PatchResultsURL(data []byte) (patched bool) { + n := 0 + for { + i := bytes.Index(data, resultsURLUTF16) + if i < 0 { + break + } + copy(data[i:], resultsRenamedUTF16) + n++ + } + if n > 0 { + return true + } + return bytes.Contains(data, resultsRenamedUTF16) +} + +// PatchResultsURLFile patches the Runner.Worker.dll under runnerRoot. +// +// When writableRoot is non-empty and differs from runnerRoot, the patched +// copy is written to writableRoot (an overlay upperdir, or a per-job copy) +// and runnerRoot is never modified — the shared runner tree must stay +// byte-identical for jobs that do not use the LAN cache. Otherwise the file +// is rewritten in place, keeping its mode. +// +// A missing assembly or an absent string is not an error: it returns false +// and the caller simply does not turn the cache on for the job. +func PatchResultsURLFile(runnerRoot, writableRoot string) (bool, error) { + src := filepath.Join(runnerRoot, WorkerDLL) + dst := src + if writableRoot != "" && writableRoot != runnerRoot { + dst = filepath.Join(writableRoot, WorkerDLL) + // An earlier patch of this same writable tree wins: it may already + // be the only copy that exists (overlay upper over a lower that was + // never touched). + if _, err := os.Stat(dst); err == nil { + src = dst + } + } + info, err := os.Stat(src) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, fmt.Errorf("stat %s: %w", src, err) + } + data, err := os.ReadFile(src) + if err != nil { + return false, fmt.Errorf("reading %s: %w", src, err) + } + if !PatchResultsURL(data) { + return false, nil + } + if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { + return false, fmt.Errorf("creating %s: %w", filepath.Dir(dst), err) + } + if err := os.WriteFile(dst, data, info.Mode().Perm()); err != nil { + return false, fmt.Errorf("writing %s: %w", dst, err) + } + return true, nil +} diff --git a/pkg/runner/resultsurl_test.go b/pkg/runner/resultsurl_test.go new file mode 100644 index 0000000..821e159 --- /dev/null +++ b/pkg/runner/resultsurl_test.go @@ -0,0 +1,154 @@ +package runner + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +func fakeDLL(parts ...[]byte) []byte { + var b []byte + b = append(b, 0x4d, 0x5a, 0x90, 0x00) // "MZ" header noise + for _, p := range parts { + b = append(b, 0x27) // #US heap length-prefix noise + b = append(b, p...) + b = append(b, 0x00) + } + return b +} + +func TestPatchResultsURL_Found(t *testing.T) { + data := fakeDLL(utf16LE("ACTIONS_RUNTIME_TOKEN"), utf16LE(ResultsURLVar), utf16LE("ACTIONS_CACHE_URL")) + size := len(data) + if !PatchResultsURL(data) { + t.Fatal("PatchResultsURL = false, want true for an assembly holding the string") + } + if len(data) != size { + t.Fatalf("patch changed the length: %d -> %d", size, len(data)) + } + if bytes.Contains(data, utf16LE(ResultsURLVar)) { + t.Error("original string still present after patch") + } + if !bytes.Contains(data, utf16LE(ResultsURLRenamed)) { + t.Error("renamed string missing after patch") + } + // Neighbours are untouched. + for _, keep := range []string{"ACTIONS_RUNTIME_TOKEN", "ACTIONS_CACHE_URL"} { + if !bytes.Contains(data, utf16LE(keep)) { + t.Errorf("patch damaged neighbouring string %q", keep) + } + } +} + +func TestPatchResultsURL_NotFound(t *testing.T) { + data := fakeDLL(utf16LE("SOMETHING_ELSE")) + // An ASCII copy is not what the CLR reads; it must not count. + data = append(data, []byte(ResultsURLVar)...) + before := append([]byte(nil), data...) + if PatchResultsURL(data) { + t.Fatal("PatchResultsURL = true for an assembly without the UTF-16 string") + } + if !bytes.Equal(before, data) { + t.Error("PatchResultsURL modified data it did not patch") + } +} + +func TestPatchResultsURL_AlreadyPatched(t *testing.T) { + data := fakeDLL(utf16LE(ResultsURLRenamed)) + if !PatchResultsURL(data) { + t.Fatal("an already-patched assembly must report true (idempotent across restarts)") + } +} + +func TestRenamedVarSameLength(t *testing.T) { + if len(ResultsURLVar) != len(ResultsURLRenamed) { + t.Fatalf("%q and %q differ in length; the #US heap entry is length-prefixed", ResultsURLVar, ResultsURLRenamed) + } +} + +func writeDLL(t *testing.T, root string, data []byte) string { + t.Helper() + p := filepath.Join(root, WorkerDLL) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, data, 0o644); err != nil { + t.Fatal(err) + } + return p +} + +func TestPatchResultsURLFile_WritesToWritableRootOnly(t *testing.T) { + lower, upper := t.TempDir(), t.TempDir() + orig := fakeDLL(utf16LE(ResultsURLVar)) + lowerPath := writeDLL(t, lower, orig) + + ok, err := PatchResultsURLFile(lower, upper) + if err != nil || !ok { + t.Fatalf("PatchResultsURLFile = %v, %v; want true, nil", ok, err) + } + got, _ := os.ReadFile(lowerPath) + if !bytes.Equal(got, orig) { + t.Fatal("the shared runner tree was modified; only the writable copy may change") + } + patched, err := os.ReadFile(filepath.Join(upper, WorkerDLL)) + if err != nil { + t.Fatalf("no patched copy in the writable root: %v", err) + } + if !bytes.Contains(patched, utf16LE(ResultsURLRenamed)) { + t.Error("writable copy is not patched") + } + + // Second call (daemon restart, same job dir) stays true. + if ok, err := PatchResultsURLFile(lower, upper); err != nil || !ok { + t.Fatalf("second PatchResultsURLFile = %v, %v; want true, nil", ok, err) + } +} + +func TestPatchResultsURLFile_InPlace(t *testing.T) { + root := t.TempDir() + p := writeDLL(t, root, fakeDLL(utf16LE(ResultsURLVar))) + if ok, err := PatchResultsURLFile(root, ""); err != nil || !ok { + t.Fatalf("PatchResultsURLFile = %v, %v; want true, nil", ok, err) + } + got, _ := os.ReadFile(p) + if !bytes.Contains(got, utf16LE(ResultsURLRenamed)) { + t.Error("in-place patch did not land") + } +} + +// TestPatchResultsURL_RealRunner runs the patch against a real +// Runner.Worker.dll when EPHEMERD_RUNNER_WORKER_DLL points at one. Run it on +// every runner bump: a new runner that stops carrying the string would +// silently turn the LAN cache off. +// +// EPHEMERD_RUNNER_WORKER_DLL=/path/to/bin/Runner.Worker.dll go test ./pkg/runner -run RealRunner -v +func TestPatchResultsURL_RealRunner(t *testing.T) { + path := os.Getenv("EPHEMERD_RUNNER_WORKER_DLL") + if path == "" { + t.Skip("set EPHEMERD_RUNNER_WORKER_DLL to a Runner.Worker.dll to check a real runner") + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if n := bytes.Count(data, utf16LE(ResultsURLVar)); n != 1 { + t.Errorf("%s holds %d UTF-16 copies of %s; the patch was verified against exactly 1", path, n, ResultsURLVar) + } + size := len(data) + if !PatchResultsURL(data) { + t.Fatal("real runner was not patched") + } + if len(data) != size || bytes.Contains(data, utf16LE(ResultsURLVar)) { + t.Fatal("patched assembly changed size or still holds the original name") + } + t.Logf("patched %s (%d bytes)", path, size) +} + +func TestPatchResultsURLFile_MissingAssembly(t *testing.T) { + ok, err := PatchResultsURLFile(t.TempDir(), t.TempDir()) + if ok || err != nil { + t.Fatalf("missing assembly = %v, %v; want false, nil (cache stays off, extraction is not failed)", ok, err) + } +} diff --git a/pkg/runtime/resultsurl.go b/pkg/runtime/resultsurl.go new file mode 100644 index 0000000..4ffde7b --- /dev/null +++ b/pkg/runtime/resultsurl.go @@ -0,0 +1,24 @@ +package runtime + +// officialRunnerRoot is where ghcr.io/actions/actions-runner (and the +// runner-ci-* images built on it) keep the runner. Jobs on those images run +// the image's own runner, not ephemerd's embedded one, so the LAN-cache patch +// has to be applied inside the container's snapshot — see +// patchSnapshotRunner. +const officialRunnerRoot = "/home/runner" + +// logResultsPatch reports how the LAN-cache runner patch went for one job. +// It never fails the job: an unpatched runner overwrites the cache URL with +// GitHub's, so the worst case is the job using GitHub's cache as before. +func (r *Runtime) logResultsPatch(id, which string) func(bool, error) { + return func(patched bool, err error) { + switch { + case err != nil: + r.cfg.Log.Warn("could not patch the runner for the LAN Actions cache; this job uses GitHub's cache", "id", id, "runner", which, "error", err) + case !patched: + r.cfg.Log.Warn("runner does not export ACTIONS_RESULTS_URL where expected (new runner version?); this job uses GitHub's cache", "id", id, "runner", which) + default: + r.cfg.Log.Info("runner patched for the LAN Actions cache", "id", id, "runner", which) + } + } +} diff --git a/pkg/runtime/resultsurl_linux.go b/pkg/runtime/resultsurl_linux.go new file mode 100644 index 0000000..837316f --- /dev/null +++ b/pkg/runtime/resultsurl_linux.go @@ -0,0 +1,32 @@ +//go:build linux + +package runtime + +import ( + "context" + "path/filepath" + + "github.com/containerd/containerd/v2/core/mount" + "github.com/containerd/containerd/v2/core/snapshots" + "github.com/ephpm/ephemerd/pkg/runner" +) + +// patchSnapshotRunner applies the LAN-cache patch (runner.PatchResultsURL) +// to the runner baked into an official runner image, inside this one +// container's writable snapshot. The write lands in the snapshot's upper +// layer, so the shared image is untouched and the patch matches whatever +// runner version the image ships. Called between NewContainer and NewTask, +// while nothing has the snapshot mounted. +func patchSnapshotRunner(ctx context.Context, sn snapshots.Snapshotter, key string) (bool, error) { + mounts, err := sn.Mounts(ctx, key) + if err != nil { + return false, err + } + var patched bool + err = mount.WithTempMount(ctx, mounts, func(root string) error { + var perr error + patched, perr = runner.PatchResultsURLFile(filepath.Join(root, officialRunnerRoot), "") + return perr + }) + return patched, err +} diff --git a/pkg/runtime/resultsurl_other.go b/pkg/runtime/resultsurl_other.go new file mode 100644 index 0000000..89ff5a2 --- /dev/null +++ b/pkg/runtime/resultsurl_other.go @@ -0,0 +1,15 @@ +//go:build !linux + +package runtime + +import ( + "context" + + "github.com/containerd/containerd/v2/core/snapshots" +) + +// patchSnapshotRunner: the LAN Actions cache is Linux-only (the scheduler +// never asks for it elsewhere), so there is nothing to patch. +func patchSnapshotRunner(context.Context, snapshots.Snapshotter, string) (bool, error) { + return false, nil +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 65d25c4..3bebe51 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -29,6 +29,7 @@ import ( "github.com/ephpm/ephemerd/pkg/networking" "github.com/ephpm/ephemerd/pkg/proxies" "github.com/ephpm/ephemerd/pkg/registrymirror" + "github.com/ephpm/ephemerd/pkg/runner" craneTarball "github.com/google/go-containerregistry/pkg/v1/tarball" ocispec "github.com/opencontainers/runtime-spec/specs-go" ) @@ -1022,6 +1023,13 @@ type CreateConfig struct { // When nil and JITConfig is set, uses the GitHub "--jitconfig" mode. // When nil and JITConfig is empty, uses the image's default CMD. Entrypoint []string + + // ResultsURL turns on ephemerd's LAN Actions cache for this job: it + // becomes the container's ACTIONS_RESULTS_URL, and the job's copy of the + // runner is patched so the runner stops overwriting it (see + // pkg/runner.PatchResultsURL). Linux only; "" leaves the job exactly as + // it was — GitHub's cache, unpatched runner. + ResultsURL string } // Create provisions an ephemeral runner environment. @@ -1117,6 +1125,15 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err for k, v := range cfg.Env { envVars = append(envVars, k+"="+v) } + // LAN Actions cache. Setting the variable is harmless on its own: an + // unpatched runner overwrites it with GitHub's address for every step, + // so if the patch below cannot be applied the job simply keeps using + // GitHub's cache. Only the patch makes the variable stick. + resultsURL := "" + if cfg.ResultsURL != "" && goruntime.GOOS == "linux" { + resultsURL = cfg.ResultsURL + envVars = append(envVars, runner.ResultsURLVar+"="+resultsURL) + } opts := []oci.SpecOpts{ oci.WithDefaultSpecForPlatform(targetPlatform), oci.WithImageConfig(img), @@ -1219,6 +1236,11 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err return nil, fmt.Errorf("preparing runner dir for %s: %w", id, err) } runnerWritableDir = writableDir + if resultsURL != "" { + // The mounted runner: patch into this job's writable layer (the + // overlay upper), never the shared tree other jobs mount. + r.logResultsPatch(id, "embedded runner")(runner.PatchResultsURLFile(r.cfg.RunnerDir, writableDir)) + } // Hyper-V isolated containers on Windows mount this host directory // into the utility VM via a VSMB share. The parent runners dir has // already been granted traverse at startup; we grant Modify scoped @@ -1267,6 +1289,7 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err BuildKit: r.cfg.BuildKit, AllowPrivileged: r.cfg.DindAllowPrivileged, RegistryMirror: r.cfg.RegistryMirror, + ResultsURL: resultsURL, Transport: resolveDindTransport(r.cfg.LinuxRuntime, goruntime.GOOS), Log: r.cfg.Log, }) @@ -1386,6 +1409,12 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err return nil, fmt.Errorf("creating container %s: %w", id, err) } + // An official runner image runs its own runner, baked into the image: + // patch it inside this container's snapshot, before the task mounts it. + if resultsURL != "" && !needsRunnerMount && snapshotter != nil { + r.logResultsPatch(id, "image runner")(patchSnapshotRunner(ctx, snapshotter, snapshotName)) + } + // Create and start the task with per-job log capture. // On Windows, cio.LogFile uses file:// URIs which runhcs rejects // (it only accepts binary:// scheme), and cio.WithStdio fails with diff --git a/pkg/scheduler/actionscache.go b/pkg/scheduler/actionscache.go new file mode 100644 index 0000000..7300028 --- /dev/null +++ b/pkg/scheduler/actionscache.go @@ -0,0 +1,78 @@ +package scheduler + +import ( + "context" + goruntime "runtime" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" +) + +// ActionsCache is the scheduler's view of the LAN Actions cache +// (pkg/proxies/actionscache). The lifecycle it drives: +// +// 1. Create: the runner is Tracked and its container gets ResultsURL. It is +// UNBOUND — every lookup misses, every save is refused. +// 2. in_progress: GitHub names the job that actually took the runner. Its +// scope is resolved from the provider's record of that job's run and +// Bound. Binding on the dispatched job instead would be wrong: same-label +// JIT runners are fungible (and org-level), so a runner started for a +// push to main can be handed a fork PR. +// 3. Teardown: the runner is Forgotten, revoking everything it was granted. +type ActionsCache interface { + Healthy() bool + ResultsURL() string + Track(runner string) + Bind(runner string, s providers.CacheScope) + Forget(runner string) +} + +// actionsCacheBindTimeout bounds the provider lookup behind a binding. A +// slow lookup only delays the job's first cache hit; it never blocks the +// event loop. +const actionsCacheBindTimeout = 30 * time.Second + +// actionsCacheFor decides whether a job about to be created gets the LAN +// cache, and if so tracks its runner and returns the ACTIONS_RESULTS_URL to +// give it. "" means the job runs exactly as it would without the feature. +func (s *Scheduler) actionsCacheFor(event providers.JobEvent, runner string) string { + ac := s.cfg.ActionsCache + if ac == nil || goruntime.GOOS != "linux" || !ac.Healthy() { + return "" + } + if _, ok := event.Provider.(providers.CacheScoper); !ok { + return "" + } + ac.Track(runner) + return ac.ResultsURL() +} + +// forgetActionsCache revokes a runner's cache access. Nil-safe. +func (s *Scheduler) forgetActionsCache(runner string) { + if ac := s.cfg.ActionsCache; ac != nil && runner != "" { + ac.Forget(runner) + } +} + +// bindActionsCache resolves and binds the cache scope of the job that took +// runner. Runs in its own goroutine from handleInProgress. Any failure +// leaves the runner unbound — cache misses, never a wider scope. +func (s *Scheduler) bindActionsCache(event providers.JobEvent, runner string) { + ac := s.cfg.ActionsCache + if ac == nil { + return + } + sc, ok := event.Provider.(providers.CacheScoper) + if !ok { + return + } + ctx, cancel := context.WithTimeout(context.Background(), actionsCacheBindTimeout) + defer cancel() + scope, err := sc.CacheScope(ctx, &event) + if err != nil { + s.cfg.Log.Warn("could not resolve the job's Actions cache scope; it runs without the LAN cache", + "runner", runner, "job_id", event.JobID, "repo", event.Repo, "error", err) + return + } + ac.Bind(runner, scope) +} diff --git a/pkg/scheduler/actionscache_test.go b/pkg/scheduler/actionscache_test.go new file mode 100644 index 0000000..3991eea --- /dev/null +++ b/pkg/scheduler/actionscache_test.go @@ -0,0 +1,157 @@ +package scheduler + +import ( + "context" + "errors" + "io" + "log/slog" + goruntime "runtime" + "sync" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" +) + +type fakeActionsCache struct { + healthy bool + mu sync.Mutex + tracked map[string]bool + bound map[string]providers.CacheScope +} + +func newFakeActionsCache() *fakeActionsCache { + return &fakeActionsCache{healthy: true, tracked: map[string]bool{}, bound: map[string]providers.CacheScope{}} +} + +func (f *fakeActionsCache) Healthy() bool { return f.healthy } +func (f *fakeActionsCache) ResultsURL() string { return "http://10.88.0.1:8089/" } +func (f *fakeActionsCache) Track(r string) { f.mu.Lock(); f.tracked[r] = true; f.mu.Unlock() } +func (f *fakeActionsCache) Forget(r string) { + f.mu.Lock() + delete(f.tracked, r) + delete(f.bound, r) + f.mu.Unlock() +} +func (f *fakeActionsCache) Bind(r string, s providers.CacheScope) { + f.mu.Lock() + f.bound[r] = s + f.mu.Unlock() +} + +// scopingProvider adds CacheScoper to the poll mock, answering from the +// event's run id so a test can tell WHICH job a binding came from. +type scopingProvider struct { + *mockProvider + err error +} + +func (p *scopingProvider) CacheScope(_ context.Context, ev *providers.JobEvent) (providers.CacheScope, error) { + if p.err != nil { + return providers.CacheScope{}, p.err + } + if ev.RunID == 1 { + return providers.CacheScope{Repo: "o/r", Write: "refs/heads/main", Reads: []string{"refs/heads/main"}}, nil + } + return providers.CacheScope{Repo: "o/r", Write: "refs/pull/9/merge", Reads: []string{"refs/pull/9/merge", "refs/heads/main"}}, nil +} + +func newCacheTestScheduler(ac ActionsCache) *Scheduler { + return &Scheduler{cfg: Config{ActionsCache: ac, Log: slog.New(slog.NewTextHandler(io.Discard, nil))}} +} + +func TestActionsCacheFor(t *testing.T) { + if goruntime.GOOS != "linux" { + t.Skip("the LAN Actions cache is Linux-only") + } + ac := newFakeActionsCache() + s := newCacheTestScheduler(ac) + scoper := &scopingProvider{mockProvider: newMockProvider("github")} + + if url := s.actionsCacheFor(providers.JobEvent{Provider: scoper}, "r1"); url == "" || !ac.tracked["r1"] { + t.Fatalf("healthy cache + scoping provider: url %q tracked %v", url, ac.tracked["r1"]) + } + if url := s.actionsCacheFor(providers.JobEvent{Provider: newMockProvider("forgejo")}, "r2"); url != "" || ac.tracked["r2"] { + t.Fatal("a provider that cannot resolve scopes got the cache") + } + ac.healthy = false + if url := s.actionsCacheFor(providers.JobEvent{Provider: scoper}, "r3"); url != "" || ac.tracked["r3"] { + t.Fatal("an unhealthy cache was turned on for a new job") + } + if url := newCacheTestScheduler(nil).actionsCacheFor(providers.JobEvent{Provider: scoper}, "r4"); url != "" { + t.Fatal("cache disabled but a URL was handed out") + } +} + +// The binding follows the job GitHub reports in in_progress — here a PR +// (run 2) that took a runner dispatched for a push to main (run 1). It must +// get the PR's scope, not main's. +func TestBindActionsCache_FollowsTheObservedJob(t *testing.T) { + ac := newFakeActionsCache() + s := newCacheTestScheduler(ac) + scoper := &scopingProvider{mockProvider: newMockProvider("github")} + + s.bindActionsCache(providers.JobEvent{Provider: scoper, RunID: 2, Repo: "r"}, "dispatched-for-main") + got := ac.bound["dispatched-for-main"] + if got.Write != "refs/pull/9/merge" { + t.Fatalf("runner bound with write %q; the PR that actually took it must not inherit main's scope", got.Write) + } +} + +func TestBindActionsCache_LookupFailureLeavesUnbound(t *testing.T) { + ac := newFakeActionsCache() + s := newCacheTestScheduler(ac) + scoper := &scopingProvider{mockProvider: newMockProvider("github"), err: errors.New("api down")} + s.bindActionsCache(providers.JobEvent{Provider: scoper, RunID: 1, Repo: "r"}, "r1") + if _, ok := ac.bound["r1"]; ok { + t.Fatal("a failed scope lookup still bound a scope") + } +} + +// countingScoper counts scope lookups, each of which is a GitHub API call. +type countingScoper struct { + *mockProvider + mu sync.Mutex + n int +} + +func (p *countingScoper) CacheScope(context.Context, *providers.JobEvent) (providers.CacheScope, error) { + p.mu.Lock() + p.n++ + p.mu.Unlock() + return providers.CacheScope{Repo: "o/r", Write: "refs/heads/main", Reads: []string{"refs/heads/main"}}, nil +} + +// Polling reports a running job on every poll, not once like a webhook. +// The binding — one API lookup — must happen on the first report only. +func TestHandleInProgress_RepeatedReportsBindOnce(t *testing.T) { + ac := newFakeActionsCache() + prov := &countingScoper{mockProvider: newMockProvider("github")} + s := New(Config{Providers: []providers.Provider{prov}, ActionsCache: ac, Log: quietLogger()}) + key := jobKey{Provider: prov.Name(), JobID: 7} + s.mu.Lock() + s.runners["r1"] = &runnerBinding{intentKey: key} + s.mu.Unlock() + + ev := providers.JobEvent{Provider: prov, Action: "in_progress", Repo: "r", JobID: 7, RunnerName: "r1"} + for i := 0; i < 5; i++ { + s.handleInProgress(ev) + } + deadline := time.Now().Add(2 * time.Second) + for { + ac.mu.Lock() + _, bound := ac.bound["r1"] + ac.mu.Unlock() + if bound || time.Now().After(deadline) { + break + } + time.Sleep(10 * time.Millisecond) + } + time.Sleep(50 * time.Millisecond) // let any duplicate binding goroutine run + prov.mu.Lock() + n := prov.n + prov.mu.Unlock() + if n != 1 { + t.Fatalf("scope looked up %d times for one job; want 1", n) + } +} diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 1a52c12..6c32714 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -111,6 +111,10 @@ type Config struct { // and finally the runtime's host-aware default. Nil-safe. RunnerImageForRepo func(repo, os string) string + // ActionsCache is ephemerd's LAN Actions cache, nil when it is off. See + // ActionsCache and actionscache.go. + ActionsCache ActionsCache + Log *slog.Logger } @@ -1991,6 +1995,7 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent // here are far cheaper than release-and-requeue. The retry queue // below is the real safety net when these run out. jobCtx, cancel := s.jobContext() + resultsURL := s.actionsCacheFor(event, claim.RunnerName) var env *runtime.RunnerEnv err = retryEnvCreate(jobCtx, log, createEnvBackoffs, func(err error) bool { return classifyErr(err) != errNonRetryable }, @@ -2009,11 +2014,14 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent // mysql:8.0 78 times in 3 days before these were passed. Provider: providerName(event), Repo: event.Repo, + // "" unless the LAN Actions cache is on for this job. + ResultsURL: resultsURL, }) return cerr }) if err != nil { log.Error("failed to create runner environment", "error", err, "error_class", classifyErr(err)) + s.forgetActionsCache(claim.RunnerName) // Remove the ghost runner since the container won't start if rmErr := event.Provider.ReleaseJob(ctx, claim); rmErr != nil { log.Warn("failed to remove ghost runner", "runner_id", claim.RunnerID, "error", rmErr) @@ -2098,6 +2106,7 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent if err := s.cfg.Runtime.Destroy(context.Background(), env); err != nil { log.Warn("failed to destroy runner environment", "error", err) } + s.forgetActionsCache(claim.RunnerName) if rj.artifactsDir != "" { artifacts.Cleanup(rj.artifactsDir, s.cfg.Log) } @@ -2144,16 +2153,24 @@ func (s *Scheduler) handleInProgress(event providers.JobEvent) { s.mu.Lock() rb, owned := s.runners[name] var intentKey jobKey + already := false if owned { + // Polling reports a running job on every poll, not once like the + // webhook does; the first report did the binding. + already = rb.bound && rb.boundKey == key rb.bound = true rb.boundKey = key intentKey = rb.intentKey } s.mu.Unlock() - if !owned { + if !owned || already { return } + // The job that ACTUALLY took this runner decides its cache scope. + if s.cfg.ActionsCache != nil { + go s.bindActionsCache(event, name) + } if intentKey != key { // The observability line for the fungibility race: GitHub gave // our runner a different job than the one it was dispatched for.