diff --git a/gitutil/command.go b/gitutil/command.go index 2477be26c..970d8ae7a 100644 --- a/gitutil/command.go +++ b/gitutil/command.go @@ -16,7 +16,7 @@ import ( // WaitDelay stops waits on output held open by git's children const WaitDelay = 10 * time.Second -// LC_ALL=C keeps git's messages in english, callers read its stderr +// LC_ALL=C keeps git's messages in english, callers match on its stderr func Command(ctx context.Context, args ...string) *exec.Cmd { cmd := exec.CommandContext(ctx, "git", args...) cmd.Env = append(os.Environ(), "LC_ALL=C") @@ -57,8 +57,17 @@ func Run(ctx context.Context, args ...string) error { } func Output(ctx context.Context, args ...string) ([]byte, error) { - var stdout, stderr bytes.Buffer + return output(Command(ctx, args...)) +} + +func OutputOffline(ctx context.Context, args ...string) ([]byte, error) { cmd := Command(ctx, args...) + cmd.Env = append(cmd.Env, "GIT_NO_LAZY_FETCH=1") + return output(cmd) +} + +func output(cmd *exec.Cmd) ([]byte, error) { + var stdout, stderr bytes.Buffer cmd.Stdout = &stdout cmd.Stderr = &stderr if err := cmd.Run(); err != nil { diff --git a/spindle/knotclient/git_read_test.go b/spindle/knotclient/git_read_test.go new file mode 100644 index 000000000..8761b4e33 --- /dev/null +++ b/spindle/knotclient/git_read_test.go @@ -0,0 +1,111 @@ +package knotclient + +import ( + "context" + "errors" + "net/http" + "strings" + "testing" +) + +func TestGetTree(t *testing.T) { + c, k := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + if r.URL.Path != "/xrpc/org.tangled.git.getTree" || q.Get("repo") != string(testRepo) || q.Get("ref") != "abc" { + t.Errorf("asked %s", r.URL) + } + switch q.Get("path") { + case ".tangled/workflows": + w.Write([]byte(`{"entries":[{"name":"ci.yaml","mode":"0100644","size":9,"lastCommit":{"hash":"x"}},{"name":"sub","mode":"0160000","size":0}]}`)) + case "bare": + http.NotFound(w, r) + default: + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":"PathNotFound","message":"path not found in repository tree"}`)) + } + }), nil) + + got, err := c.GetTree(context.Background(), k, testRepo, "abc", ".tangled/workflows") + if err != nil { + t.Fatal(err) + } + if len(got) != 2 || got[0] != (TreeEntry{Name: "ci.yaml", Mode: "0100644", Size: 9}) || got[1].Mode != "0160000" { + t.Errorf("entries = %+v", got) + } + if _, err := c.GetTree(context.Background(), k, testRepo, "abc", "nope"); !errors.Is(err, ErrPathNotFound) || errors.Is(err, ErrNotFound) { + t.Errorf("missing dir err = %v, want only ErrPathNotFound", err) + } + if _, err := c.GetTree(context.Background(), k, testRepo, "abc", "bare"); err == nil || errors.Is(err, ErrPathNotFound) || errors.Is(err, ErrNotFound) { + t.Errorf("bare 404 err = %v, want a plain error", err) + } +} + +func TestGetTreeOfAGoneRepoIsErrNotFound(t *testing.T) { + c, k := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":"RepoNotFound"}`)) + }), nil) + if _, err := c.GetTree(context.Background(), k, testRepo, "abc", "x"); !errors.Is(err, ErrNotFound) { + t.Errorf("err = %v, want ErrNotFound", err) + } +} + +func TestGetBlob(t *testing.T) { + c, k := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + if r.URL.Path != "/xrpc/org.tangled.git.getBlob" || q.Get("repo") != string(testRepo) || q.Get("ref") != "abc" { + t.Errorf("asked %s", r.URL) + } + switch q.Get("path") { + case "ci.yaml": + w.Write([]byte("engine: nixery\n")) + case "big": + w.Write([]byte(strings.Repeat("x", maxResponseBody+1))) + case "limit": + w.Write([]byte(strings.Repeat("x", maxResponseBody))) + default: + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":"FileNotFound"}`)) + } + }), nil) + + got, err := c.GetBlob(context.Background(), k, testRepo, "abc", "ci.yaml") + if err != nil || string(got) != "engine: nixery\n" { + t.Errorf("blob = %q, %v", got, err) + } + if got, err := c.GetBlob(context.Background(), k, testRepo, "abc", "limit"); err != nil || len(got) != maxResponseBody { + t.Errorf("blob at the cap = %d bytes, %v, want it served", len(got), err) + } + if _, err := c.GetBlob(context.Background(), k, testRepo, "abc", "big"); !errors.Is(err, ErrBlobTooLarge) { + t.Errorf("blob past the cap err = %v, want ErrBlobTooLarge", err) + } + if _, err := c.GetBlob(context.Background(), k, testRepo, "abc", "ghost"); !errors.Is(err, ErrPathNotFound) { + t.Errorf("missing file err = %v, want ErrPathNotFound", err) + } +} + +func TestGetTagAndRefNotFound(t *testing.T) { + c, k := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + switch { + case r.URL.Path == "/xrpc/org.tangled.git.getTag" && q.Get("tag") == "v1": + w.Write([]byte(`{"name":"v1","oid":"abc","message":"m","author":{}}`)) + case r.URL.Path == "/xrpc/org.tangled.git.getTag": + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":"RefNotFound"}`)) + default: + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":"BranchNotFound"}`)) + } + }), nil) + + if oid, err := c.GetTag(context.Background(), k, testRepo, "v1"); err != nil || oid != "abc" { + t.Errorf("tag = %q, %v", oid, err) + } + if _, err := c.GetTag(context.Background(), k, testRepo, "v2"); !errors.Is(err, ErrRefNotFound) { + t.Errorf("missing tag err = %v, want ErrRefNotFound", err) + } + if _, err := c.GetBranch(context.Background(), k, testRepo, "nope"); !errors.Is(err, ErrRefNotFound) { + t.Errorf("missing branch err = %v, want ErrRefNotFound", err) + } +} diff --git a/spindle/knotclient/knotclient.go b/spindle/knotclient/knotclient.go index 27bb6b983..9178e1899 100644 --- a/spindle/knotclient/knotclient.go +++ b/spindle/knotclient/knotclient.go @@ -36,6 +36,9 @@ var ( ErrTooManyRecords = errors.New("more records than spindle pages through") ErrNotFound = errors.New("record not found on the knot") ErrNoKnot = errors.New("repo did document names no knot") + ErrPathNotFound = errors.New("path not found on the knot") + ErrBlobTooLarge = errors.New("blob is larger than spindle reads") + ErrRefNotFound = errors.New("branch or tag not found on the knot") ) type Knot struct { @@ -152,7 +155,6 @@ type RecordVersion struct { Editor syntax.DID } -// a version without a cid is a removal, so it's skipped func (c *Client) ListRecordVersions(ctx context.Context, k Knot, repo syntax.DID, collection syntax.NSID, rkey string) ([]RecordVersion, error) { uri := fmt.Sprintf("at://%s/%s/%s", repo, collection, rkey) var versions []RecordVersion @@ -204,6 +206,64 @@ func (c *Client) GetBranch(ctx context.Context, k Knot, repo syntax.DID, name st return out.Hash, nil } +type TreeEntry struct { + Name string + Mode string // octal, as the knot sends it + Size int64 +} + +// GetTree gives ErrPathNotFound when ref has no such directory +func (c *Client) GetTree(ctx context.Context, k Knot, repo syntax.DID, ref, path string) ([]TreeEntry, error) { + q := url.Values{} + q.Set("repo", repo.String()) + q.Set("ref", ref) + q.Set("path", path) + var out org_tangled.GitGetTree_Output + if err := c.get(ctx, k, org_tangled.GitGetTreeNSID, q, &out); err != nil { + return nil, err + } + entries := make([]TreeEntry, 0, len(out.Entries)) + for _, e := range out.Entries { + if e != nil { + entries = append(entries, TreeEntry{Name: e.Name, Mode: e.Mode, Size: e.Size}) + } + } + return entries, nil +} + +// GetBlob gives ErrPathNotFound when ref has no such file, ErrBlobTooLarge past the response cap +func (c *Client) GetBlob(ctx context.Context, k Knot, repo syntax.DID, ref, path string) ([]byte, error) { + q := url.Values{} + q.Set("repo", repo.String()) + q.Set("ref", ref) + q.Set("path", path) + body, err := c.fetch(ctx, k, org_tangled.GitGetBlobNSID, q) + if err != nil { + return nil, err + } + defer body.Close() + data, err := io.ReadAll(io.LimitReader(body, maxResponseBody+1)) + if err != nil { + return nil, err + } + if len(data) > maxResponseBody { + return nil, fmt.Errorf("%s of %s: %w", path, repo, ErrBlobTooLarge) + } + return data, nil +} + +// the oid of the commit the tag points at, peeled when the tag is annotated +func (c *Client) GetTag(ctx context.Context, k Knot, repo syntax.DID, name string) (string, error) { + q := url.Values{} + q.Set("repo", repo.String()) + q.Set("tag", name) + var out org_tangled.GitGetTag_Output + if err := c.get(ctx, k, org_tangled.GitGetTagNSID, q, &out); err != nil { + return "", err + } + return out.Oid, nil +} + func (c *Client) base(k Knot) string { if k.Scheme != "" { return k.Scheme + "://" + k.Host @@ -226,31 +286,48 @@ func hostScheme(dev bool, host string) string { // no service auth yet, so private repos would need it added here func (c *Client) get(ctx context.Context, k Knot, method string, q url.Values, out any) error { + body, err := c.fetch(ctx, k, method, q) + if err != nil { + return err + } + defer body.Close() + return json.NewDecoder(io.LimitReader(body, maxResponseBody)).Decode(out) +} + +func (c *Client) fetch(ctx context.Context, k Knot, method string, q url.Values) (io.ReadCloser, error) { u := fmt.Sprintf("%s/xrpc/%s?%s", c.base(k), method, q.Encode()) req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) if err != nil { - return err + return nil, err } resp, err := c.http.Do(req) if err != nil { - return err + return nil, err + } + if resp.StatusCode == http.StatusOK { + return resp.Body, nil } defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody)) - // only the knot naming the miss counts, a bare 404 from a proxy in front of it - // would otherwise read as the repo being gone and wipe its secrets - if isNotFound(body) { - return fmt.Errorf("%s: %w", method, ErrNotFound) - } - return fmt.Errorf("%s on %s answered %s: %s", method, k.Host, resp.Status, body) + body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody)) + // only the knot naming the miss counts, a bare 404 from a proxy + // would look like the repo is gone and wipe its secrets + switch errorName(body) { + case "RecordNotFound", "RepoNotFound": + return nil, fmt.Errorf("%s: %w", method, ErrNotFound) + case "BranchNotFound", "RefNotFound": + return nil, fmt.Errorf("%s: %w", method, ErrRefNotFound) + case "PathNotFound", "FileNotFound": + return nil, fmt.Errorf("%s: %w", method, ErrPathNotFound) } - return json.NewDecoder(io.LimitReader(resp.Body, maxResponseBody)).Decode(out) + return nil, fmt.Errorf("%s on %s answered %s: %s", method, k.Host, resp.Status, body) } -func isNotFound(body []byte) bool { +func errorName(body []byte) string { var e struct { Error string `json:"error"` } - return json.Unmarshal(body, &e) == nil && (e.Error == "RecordNotFound" || e.Error == "RepoNotFound") + if json.Unmarshal(body, &e) != nil { + return "" + } + return e.Error } diff --git a/spindle/server.go b/spindle/server.go index 44ef69c20..fc49b14df 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -13,7 +13,6 @@ import ( "maps" "net/http" "net/url" - pathpkg "path" "path/filepath" "regexp" "sort" @@ -533,7 +532,7 @@ func (s *Spindle) Start(ctx context.Context) error { shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), serverShutdownTimeout) shutdownErr := server.Shutdown(shutdownCtx) cancelShutdown() - // notify work can outlive the handler, so wait for it before the run ends + // give notify work a moment before the run ends, since it can outlast the handler notifyDone := make(chan struct{}) go func() { s.notifyWork.Wait() @@ -805,7 +804,7 @@ func Run(ctx context.Context) error { return s.Start(ctx) } -// no write timeout since log streams run for hours, and a websocket upgrade clears the read deadline +// no write timeout, log streams run for hours and would get cut off func newHTTPServer(addr string, handler http.Handler) *http.Server { return &http.Server{ Addr: addr, @@ -929,8 +928,7 @@ func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) return s.buildTriggerRepo(ctx, repo) } - // git sync must not point at an arbitrary endpoint, so the knot comes from the did doc - // and the owner's pds has to confirm what that knot says + // the knot comes from the did doc, not the caller, or the source could point anywhere k, err := s.knots.KnotOf(ctx, repoDid) if err != nil { return nil, fmt.Errorf("verify sourceRepo %s: %w", repoDid, err) @@ -998,12 +996,11 @@ func (s *Spindle) runPipeline(ctx context.Context, pipelineId models.PipelineId, return "", fmt.Errorf("creating pipeline: %w", err) } err = s.processPipeline(ctx, repoDid, tpl, record, sourceRepo) - // a pause or wipe can land during the checkout, so either error means nothing to enqueue + // both errors mean nothing to enqueue, since a pause or wipe can land mid checkout if errors.Is(err, db.ErrOwnerPaused) || errors.Is(err, db.ErrNotServed) { l.Info("nothing enqueued, the owner is paused or the repo is gone", "repo", repoDid) return "", nil } - // the id is only real once the pipeline row exists if err != nil { return "", err } @@ -1040,7 +1037,7 @@ func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, } else if ban != nil { return "", fmt.Errorf("repo %s is banned", repoDid) } - // checked before the checkout, so a paused owner gets an error and not an empty match + // checked before the checkout, otherwise a paused owner matches nothing instead of erroring if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil { return "", fmt.Errorf("checking admission: %w", err) } else if gate.Paused { @@ -1152,11 +1149,7 @@ func (s *Spindle) ListWorkflowDefinitions(ctx context.Context, repoDid syntax.DI } func (s *Spindle) loadWorkflowDefinitions(ctx context.Context, repoDid syntax.DID, ref string) ([]*models.WorkflowDefinition, error) { - repoCloneURI, repoPath, _, err := s.resolveCheckout(ctx, repoDid, "") - if err != nil { - return nil, err - } - rawPipeline, commit, err := s.loadPipeline(ctx, repoCloneURI, repoPath, ref) + rawPipeline, commit, err := s.readWorkflowDir(ctx, repoDid, ref) if err != nil { return nil, fmt.Errorf("loading pipeline: %w", err) } @@ -1172,7 +1165,7 @@ func (s *Spindle) loadWorkflowDefinitions(ctx context.Context, repoDid syntax.DI } func (s *Spindle) DescribeWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - // anyone can ask, so the source repo must be one we already serve + // anyone can ask, the source has to be a repo we already serve if sourceRepo != "" && sourceRepo != repoDid { if _, err := s.db.GetRepoByDid(sourceRepo); errors.Is(err, sql.ErrNoRows) { return nil, xrpc.ErrSourceRepoNotServed @@ -1186,12 +1179,11 @@ func (s *Spindle) DescribeWorkflowDefinition(ctx context.Context, repoDid syntax } func (s *Spindle) loadWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - repoCloneUri, repoPath, _, err := s.resolveCheckout(ctx, repoDid, sourceRepo) - if err != nil { - return nil, err + readFrom := repoDid + if sourceRepo != "" { + readFrom = sourceRepo } - - rawPipeline, _, err := s.loadPipeline(ctx, repoCloneUri, repoPath, sha) + rawPipeline, _, err := s.readWorkflowDir(ctx, readFrom, sha) if err != nil { return nil, fmt.Errorf("loading pipeline: %w", err) } @@ -1250,11 +1242,11 @@ var ( // a 4xx means the knot won't serve that commit, a retry gets the same answer unless it was busy func refusedByKnot(err error) bool { msg := err.Error() - // upload-pack reports an unknown sha as a 200, so there's no status code to check + // upload-pack answers 200 for an unknown sha, which leaves only the message to match if strings.Contains(msg, "not our ref") || strings.Contains(msg, "unadvertised object") { return true } - // a plain 404 on info/refs only shows up in the message text + // info/refs 404s only show up in the message text if strings.Contains(msg, "fatal: repository '") && strings.Contains(msg, "' not found") { return true } @@ -1286,29 +1278,8 @@ func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev strin return nil, "", fmt.Errorf("resolving checked out commit: %w", err) } commit := strings.TrimSpace(string(head)) - - // a committed .tangled symlink could point the checkout at files on this host - listing, err := gitutil.Output(ctx, "-C", repoPath, "ls-tree", "-z", commit, workflow.WorkflowDir+"/") - if err != nil { - return nil, "", fmt.Errorf("listing %s: %w", workflow.WorkflowDir, err) - } - var rawPipeline workflow.RawPipeline - for _, entry := range strings.Split(string(listing), "\x00") { - meta, path, ok := strings.Cut(entry, "\t") - fields := strings.Fields(meta) - if !ok || len(fields) != 3 || (fields[0] != "100644" && fields[0] != "100755") { - continue - } - if len(rawPipeline) >= maxWorkflowFiles { - return nil, "", fmt.Errorf("%w: more than %d files under %s", errTooManyWorkflowFiles, maxWorkflowFiles, workflow.WorkflowDir) - } - contents, err := gitutil.Output(ctx, "-C", repoPath, "cat-file", "blob", fields[2]) - if err != nil { - return nil, "", fmt.Errorf("reading raw content of '%s': %w", path, err) - } - rawPipeline = append(rawPipeline, workflow.RawWorkflow{Name: pathpkg.Base(path), Contents: contents}) - } - return rawPipeline, commit, nil + rawPipeline, err := readWorkflowTree(ctx, gitutil.Output, repoPath, commit) + return rawPipeline, commit, err } func (s *Spindle) newRepoPath(repo syntax.DID) string { @@ -1371,7 +1342,7 @@ func (s *Spindle) StartJobWorkers(ctx context.Context) { s.l.Error("failed to dequeue job", "error", err) } if job == nil { - // a failed dequeue can leave jobs queued with nothing to wake us, so retry on a timer + // polls because a failed dequeue can leave jobs queued with nobody to wake us var retry <-chan time.Time if err != nil { retry = time.After(dequeueRetryDelay) @@ -1566,7 +1537,7 @@ func (s *Spindle) idTokenMinter(job *db.JobRow, trustedSource bool) func(time.Du func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, record *models.PipelineRecord, sourceRepo *tangled.Pipeline_TriggerRepo) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) - // only the insert holds the repo lock, so a wipe can't land in the middle of it + // only the insert holds the repo lock, which keeps a wipe from landing halfway through err := func() error { defer s.lockRepo(repoDid).Unlock() return s.db.EnqueuePipeline(s.rootCtx, repoDid.String(), record, sourceRepo, tpl, traceparent, tracestate, s.n, db.Admitted(&s.cfg.Server)) diff --git a/spindle/workflow_defs.go b/spindle/workflow_defs.go index 03734786f..69a4e3b67 100644 --- a/spindle/workflow_defs.go +++ b/spindle/workflow_defs.go @@ -14,15 +14,15 @@ import ( ) const ( - // commits never change, refs can move, so only refs get the short ttl + // only refs get the short ttl because commits don't move but refs do workflowDefCommitTTL = 10 * time.Minute workflowDefRefTTL = 10 * time.Second workflowDefCacheSize = 512 - workflowDefLoadsPerSecond = 1 - workflowDefLoadBurst = 5 + workflowDefLoadsPerSecond = 5 + workflowDefLoadBurst = 20 ) -// every miss is a git fetch under the same lock pushes take, so cache the answers +// a miss costs the knot a tree read and a blob per workflow type workflowDefCache struct { once sync.Once mu sync.Mutex @@ -37,7 +37,9 @@ func (c *workflowDefCache) init() { c.limits = expirable.NewLRU[syntax.DID, *rate.Limiter](workflowDefCacheSize, nil, time.Minute) } -func (c *workflowDefCache) allow(repo syntax.DID) bool { +// only knot reads spend the budget, local clone reads are free +func (c *workflowDefCache) admit(repo syntax.DID) error { + c.once.Do(c.init) c.mu.Lock() defer c.mu.Unlock() limiter, ok := c.limits.Get(repo) @@ -45,27 +47,26 @@ func (c *workflowDefCache) allow(repo syntax.DID) bool { limiter = rate.NewLimiter(workflowDefLoadsPerSecond, workflowDefLoadBurst) c.limits.Add(repo, limiter) } - return limiter.Allow() + if !limiter.Allow() { + return xrpc.ErrDefinitionsBusy + } + return nil } func (c *workflowDefCache) list(repo syntax.DID, ref string, load func() ([]*models.WorkflowDefinition, error)) ([]*models.WorkflowDefinition, error) { c.once.Do(c.init) - return cached(c, c.byRef, repo, repo.String()+"\x00"+ref, load) + return cached(c.byRef, repo.String()+"\x00"+ref, load) } func (c *workflowDefCache) describe(repo, source syntax.DID, sha string, load func() (*tangled.CiDescribeWorkflowDefinition_Output, error)) (*tangled.CiDescribeWorkflowDefinition_Output, error) { c.once.Do(c.init) - return cached(c, c.byCommit, repo, repo.String()+"\x00"+source.String()+"\x00"+sha, load) + return cached(c.byCommit, repo.String()+"\x00"+source.String()+"\x00"+sha, load) } -func cached[V any](c *workflowDefCache, lru *expirable.LRU[string, V], repo syntax.DID, key string, load func() (V, error)) (V, error) { +func cached[V any](lru *expirable.LRU[string, V], key string, load func() (V, error)) (V, error) { if v, ok := lru.Get(key); ok { return v, nil } - if !c.allow(repo) { - var zero V - return zero, xrpc.ErrDefinitionsBusy - } v, err := load() if err == nil { lru.Add(key, v) diff --git a/spindle/workflow_defs_test.go b/spindle/workflow_defs_test.go index a5fe6761e..af5eee903 100644 --- a/spindle/workflow_defs_test.go +++ b/spindle/workflow_defs_test.go @@ -5,6 +5,8 @@ import ( "fmt" "testing" + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/spindle/models" "tangled.org/core/spindle/xrpc" ) @@ -12,13 +14,15 @@ import ( func TestWorkflowDefinitionsAreCachedAndRateLimitedPerRepo(t *testing.T) { var c workflowDefCache loads := 0 - load := func() ([]*models.WorkflowDefinition, error) { - loads++ - return []*models.WorkflowDefinition{{ID: "ci"}}, nil + knotRead := func(repo syntax.DID) func() ([]*models.WorkflowDefinition, error) { + return func() ([]*models.WorkflowDefinition, error) { + loads++ + return []*models.WorkflowDefinition{{ID: "ci"}}, c.admit(repo) + } } for range 20 { - if _, err := c.list("did:plc:repo", "HEAD", load); err != nil { + if _, err := c.list("did:plc:repo", "HEAD", knotRead("did:plc:repo")); err != nil { t.Fatal(err) } } @@ -26,18 +30,31 @@ func TestWorkflowDefinitionsAreCachedAndRateLimitedPerRepo(t *testing.T) { t.Fatalf("%d loads for 20 reads of the same ref, want 1", loads) } - // every uncached ref is a git fetch, so this trips the busy limit + // distinct refs skip the cache and each one is a knot read var busy error - for i := range 10 { - if _, err := c.list("did:plc:repo", fmt.Sprintf("refs/heads/b%d", i), load); err != nil { + for i := range 3 * workflowDefLoadBurst { + if _, err := c.list("did:plc:repo", fmt.Sprintf("refs/heads/b%d", i), knotRead("did:plc:repo")); err != nil { busy = err break } } if !errors.Is(busy, xrpc.ErrDefinitionsBusy) { - t.Fatalf("10 uncached reads of one repo in a row ended with %v, want ErrDefinitionsBusy", busy) + t.Fatalf("%d uncached reads of one repo in a row ended with %v, want ErrDefinitionsBusy", 3*workflowDefLoadBurst, busy) } - if _, err := c.list("did:plc:otherrepo", "HEAD", load); err != nil { + if _, err := c.list("did:plc:otherrepo", "HEAD", knotRead("did:plc:otherrepo")); err != nil { t.Fatalf("another repo was refused after the first one got busy: %v", err) } } + +func TestWorkflowDefinitionRateLimitOnlyCountsKnotReads(t *testing.T) { + var c workflowDefCache + local := func() ([]*models.WorkflowDefinition, error) { return nil, nil } + for i := range 10 * workflowDefLoadBurst { + if _, err := c.list("did:plc:repo", fmt.Sprintf("%040d", i), local); err != nil { + t.Fatalf("local read %d was limited: %v", i, err) + } + } + if err := c.admit("did:plc:repo"); err != nil { + t.Fatalf("local reads used up the knot budget: %v", err) + } +} diff --git a/spindle/workflow_read.go b/spindle/workflow_read.go new file mode 100644 index 000000000..53d924084 --- /dev/null +++ b/spindle/workflow_read.go @@ -0,0 +1,152 @@ +package spindle + +import ( + "context" + "errors" + "fmt" + pathpkg "path" + "strings" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "tangled.org/core/gitutil" + "tangled.org/core/spindle/db" + "tangled.org/core/spindle/knotclient" + "tangled.org/core/workflow" +) + +// no fetch here, it uses a clone that already has the commit or asks the knot +func (s *Spindle) readWorkflowDir(ctx context.Context, repoDid syntax.DID, ref string) (workflow.RawPipeline, string, error) { + repo, err := s.db.GetRepoByDid(repoDid) + if err != nil { + return nil, "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) + } + k := knotclient.Knot{Host: repo.Knot} + + admitted := false + admit := func() error { + if admitted { + return nil + } + admitted = true + return s.workflowDefs.admit(repoDid) + } + + commit := ref + if !gitutil.Rev(ref).IsObjectID() { + if err := admit(); err != nil { + return nil, "", err + } + if commit, err = s.resolveWorkflowRef(ctx, *repo, ref); err != nil { + return nil, "", err + } + } + + raw, ok, err := s.readLocalWorkflows(ctx, s.newRepoPath(repoDid), commit) + if ok || err != nil { + return raw, commit, err + } + if err := admit(); err != nil { + return nil, "", err + } + raw, err = s.readKnotWorkflows(ctx, k, repoDid, commit) + return raw, commit, err +} + +func (s *Spindle) resolveWorkflowRef(ctx context.Context, repo db.Repo, ref string) (string, error) { + if ref == "HEAD" { + _, hash, err := s.getRepoDefaultBranch(ctx, repo) + return hash, err + } + k := knotclient.Knot{Host: repo.Knot} + var hash string + var err error + switch tag, isTag := strings.CutPrefix(ref, "refs/tags/"); { + case isTag: + hash, err = s.knots.GetTag(ctx, k, repo.RepoDid, tag) + case strings.HasPrefix(ref, "refs/heads/"): + hash, err = s.knots.GetBranch(ctx, k, repo.RepoDid, strings.TrimPrefix(ref, "refs/heads/")) + default: + // a bare name checks branches before tags, like git does + if hash, err = s.knots.GetBranch(ctx, k, repo.RepoDid, ref); errors.Is(err, knotclient.ErrRefNotFound) { + hash, err = s.knots.GetTag(ctx, k, repo.RepoDid, ref) + } + } + if err != nil { + return "", fmt.Errorf("resolving %s of %s: %w", ref, repo.RepoDid, err) + } + if hash == "" { + return "", fmt.Errorf("%s of %s has no commit hash", ref, repo.RepoDid) + } + return hash, nil +} + +// no lock, a fetch replacing the clone only makes git fail, which falls back to the knot +func (s *Spindle) readLocalWorkflows(ctx context.Context, repoPath, commit string) (raw workflow.RawPipeline, ok bool, err error) { + if _, err := gitutil.OutputOffline(ctx, "-C", repoPath, "cat-file", "-e", commit+"^{commit}"); err != nil { + return nil, false, nil + } + raw, err = readWorkflowTree(ctx, gitutil.OutputOffline, repoPath, commit) + if errors.Is(err, errTooManyWorkflowFiles) { + return nil, true, err + } + return raw, err == nil, nil +} + +func (s *Spindle) readKnotWorkflows(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, commit string) (workflow.RawPipeline, error) { + entries, err := s.knots.GetTree(ctx, k, repoDid, commit, workflow.WorkflowDir) + if errors.Is(err, knotclient.ErrPathNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("listing %s: %w", workflow.WorkflowDir, err) + } + var raw workflow.RawPipeline + for _, entry := range entries { + if !isRegularFileMode(entry.Mode) { + continue + } + if len(raw) >= maxWorkflowFiles { + return nil, fmt.Errorf("%w: more than %d files under %s", errTooManyWorkflowFiles, maxWorkflowFiles, workflow.WorkflowDir) + } + contents, err := s.knots.GetBlob(ctx, k, repoDid, commit, workflow.WorkflowDir+"/"+entry.Name) + if err != nil { + return nil, fmt.Errorf("reading raw content of '%s': %w", entry.Name, err) + } + raw = append(raw, workflow.RawWorkflow{Name: entry.Name, Contents: contents}) + } + return raw, nil +} + +// the knot pads modes to seven digits, git prints six +func isRegularFileMode(mode string) bool { + mode = strings.TrimLeft(mode, "0") + return mode == "100644" || mode == "100755" +} + +type gitOutput func(ctx context.Context, args ...string) ([]byte, error) + +// a symlink in .tangled could point reads at files on this host +func readWorkflowTree(ctx context.Context, git gitOutput, repoPath, commit string) (workflow.RawPipeline, error) { + listing, err := git(ctx, "-C", repoPath, "ls-tree", "-z", commit, workflow.WorkflowDir+"/") + if err != nil { + return nil, fmt.Errorf("listing %s: %w", workflow.WorkflowDir, err) + } + var raw workflow.RawPipeline + for _, entry := range strings.Split(string(listing), "\x00") { + meta, path, ok := strings.Cut(entry, "\t") + fields := strings.Fields(meta) + if !ok || len(fields) != 3 || !isRegularFileMode(fields[0]) { + continue + } + if len(raw) >= maxWorkflowFiles { + return nil, fmt.Errorf("%w: more than %d files under %s", errTooManyWorkflowFiles, maxWorkflowFiles, workflow.WorkflowDir) + } + contents, err := git(ctx, "-C", repoPath, "cat-file", "blob", fields[2]) + if err != nil { + return nil, fmt.Errorf("reading raw content of '%s': %w", path, err) + } + raw = append(raw, workflow.RawWorkflow{Name: pathpkg.Base(path), Contents: contents}) + } + return raw, nil +} diff --git a/spindle/workflow_read_test.go b/spindle/workflow_read_test.go new file mode 100644 index 000000000..21a58e85f --- /dev/null +++ b/spindle/workflow_read_test.go @@ -0,0 +1,402 @@ +package spindle + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/bluesky-social/indigo/atproto/identity" + "github.com/bluesky-social/indigo/atproto/syntax" + + "tangled.org/core/gitutil" + "tangled.org/core/spindle/config" + "tangled.org/core/spindle/db" + "tangled.org/core/spindle/knotclient" +) + +const defsWorkflow = "engine: nixery\n" + +type defsFile struct{ name, mode, body string } + +type defsKnot struct { + srv *httptest.Server + + mu sync.Mutex + branches map[string]string + tags map[string]string + trees map[string][]defsFile + requests []string +} + +func newDefsKnot(t *testing.T) *defsKnot { + t.Helper() + k := &defsKnot{branches: map[string]string{}, tags: map[string]string{}, trees: map[string][]defsFile{}} + k.srv = httptest.NewServer(http.HandlerFunc(k.serve)) + t.Cleanup(k.srv.Close) + return k +} + +func (k *defsKnot) host() string { + u, _ := url.Parse(k.srv.URL) + return u.Host +} + +func (k *defsKnot) setBranch(repo syntax.DID, name, commit string) { + k.mu.Lock() + defer k.mu.Unlock() + k.branches[repo.String()+"|"+name] = commit +} + +func (k *defsKnot) setTag(repo syntax.DID, name, commit string) { + k.mu.Lock() + defer k.mu.Unlock() + k.tags[repo.String()+"|"+name] = commit +} + +func (k *defsKnot) setTree(repo syntax.DID, commit string, files ...defsFile) { + k.mu.Lock() + defer k.mu.Unlock() + k.trees[repo.String()+"|"+commit] = files +} + +func (k *defsKnot) asked() []string { + k.mu.Lock() + defer k.mu.Unlock() + return append([]string(nil), k.requests...) +} + +func (k *defsKnot) count(method string) int { + n := 0 + for _, r := range k.asked() { + if strings.HasPrefix(r, method+" ") { + n++ + } + } + return n +} + +func (k *defsKnot) serve(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + method := strings.TrimPrefix(r.URL.Path, "/xrpc/") + k.mu.Lock() + defer k.mu.Unlock() + k.requests = append(k.requests, method+" "+q.Get("ref")+q.Get("name")+q.Get("tag")+" "+q.Get("path")) + fail := func(status int, name string) { + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]string{"error": name}) + } + repo := q.Get("repo") + switch method { + case "com.atproto.repo.getRecord": + json.NewEncoder(w).Encode(map[string]any{"value": map[string]any{"defaultBranch": "main"}}) + case "org.tangled.git.getBranch": + hash, ok := k.branches[repo+"|"+q.Get("name")] + if !ok { + fail(http.StatusNotFound, "BranchNotFound") + return + } + json.NewEncoder(w).Encode(map[string]string{"hash": hash}) + case "org.tangled.git.getTag": + oid, ok := k.tags[repo+"|"+q.Get("tag")] + if !ok { + fail(http.StatusNotFound, "RefNotFound") + return + } + json.NewEncoder(w).Encode(map[string]string{"oid": oid}) + case "org.tangled.git.getTree": + files, ok := k.trees[repo+"|"+q.Get("ref")] + if !ok || q.Get("path") != ".tangled/workflows" { + fail(http.StatusNotFound, "PathNotFound") + return + } + entries := []map[string]any{} + for _, f := range files { + entries = append(entries, map[string]any{"name": f.name, "mode": f.mode, "size": len(f.body)}) + } + json.NewEncoder(w).Encode(map[string]any{"entries": entries}) + case "org.tangled.git.getBlob": + for _, f := range k.trees[repo+"|"+q.Get("ref")] { + if ".tangled/workflows/"+f.name == q.Get("path") { + w.Write([]byte(f.body)) + return + } + } + fail(http.StatusNotFound, "FileNotFound") + default: + fail(http.StatusNotFound, "MethodNotImplemented") + } +} + +func newDefsSpindle(t *testing.T, knot *defsKnot, repos ...syntax.DID) *Spindle { + t.Helper() + ctx := context.Background() + d, err := db.Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { d.Close() }) + for i, repo := range repos { + if err := d.AddRepo(db.Repo{Knot: knot.host(), Owner: "did:plc:owner", Rkey: syntax.RecordKey(fmt.Sprintf("repo%d", i)), RepoDid: repo}); err != nil { + t.Fatal(err) + } + } + cfg := &config.Config{} + cfg.Server.RepoDir = filepath.Join(t.TempDir(), "repos") + return &Spindle{db: d, cfg: cfg, knots: knotclient.New(identity.NewMockDirectory(), http.DefaultClient, true)} +} + +func regular(name string) defsFile { return defsFile{name, "0100644", defsWorkflow} } + +// the origin is a dead path on purpose, like a clone whose remote is gone +func cloneWithCommit(t *testing.T, s *Spindle, repo syntax.DID) (src, commit string) { + t.Helper() + src = newSourceRepo(t, "sha1") + writeTestFile(t, filepath.Join(src, ".tangled/workflows/local.yaml"), defsWorkflow) + commitEverything(t, src) + commit = gitInDir(t, src, "rev-parse", "HEAD") + path := s.newRepoPath(repo) + if err := gitutil.SparseSync(context.Background(), src, path, commit, sparseWorkflowDir); err != nil { + t.Fatal(err) + } + gitInDir(t, path, "remote", "set-url", "origin", filepath.Join(t.TempDir(), "gone")) + return src, commit +} + +func definitionNames(t *testing.T, s *Spindle, repo syntax.DID, ref string) (names []string, commits []string) { + t.Helper() + defs, err := s.ListWorkflowDefinitions(context.Background(), repo, ref) + if err != nil { + t.Fatal(err) + } + for _, d := range defs { + names = append(names, d.Name) + commits = append(commits, d.Source.File.Commit) + } + return names, commits +} + +func TestDefinitionMissReadsTheLocalCloneWithoutTheKnotOrAFetch(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + _, commit := cloneWithCommit(t, s, repo) + + names, commits := definitionNames(t, s, repo, commit) + if len(names) != 1 || names[0] != "local.yaml" || commits[0] != commit { + t.Fatalf("definitions = %v at %v, want local.yaml at %s", names, commits, commit) + } + if got := knot.asked(); len(got) != 0 { + t.Fatalf("a clone that had the commit still cost the knot %v", got) + } +} + +func TestDefinitionMissWithoutTheCommitLocallyReadsTheKnot(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + cloneWithCommit(t, s, repo) + other := strings.Repeat("b", 40) + knot.setTree(repo, other, regular("remote.yaml")) + + names, commits := definitionNames(t, s, repo, other) + if len(names) != 1 || names[0] != "remote.yaml" || commits[0] != other { + t.Fatalf("definitions = %v at %v, want remote.yaml at %s", names, commits, other) + } +} + +func TestDefinitionMissWithoutACloneReadsTheKnotAndNeverRunsGit(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + commit := strings.Repeat("a", 40) + knot.setTree(repo, commit, regular("ci.yaml"), regular("deploy.yaml")) + t.Setenv("PATH", "") + + names, commits := definitionNames(t, s, repo, commit) + if strings.Join(names, ",") != "ci.yaml,deploy.yaml" || commits[0] != commit { + t.Fatalf("definitions = %v at %v", names, commits) + } + if knot.count("org.tangled.git.getTree") != 1 || knot.count("org.tangled.git.getBlob") != 2 { + t.Fatalf("knot saw %v, want one getTree and a getBlob per file", knot.asked()) + } + if _, err := os.Stat(s.newRepoPath(repo)); !os.IsNotExist(err) { + t.Fatalf("a read made a clone dir: %v", err) + } +} + +func TestDefinitionKnotReadSkipsEverythingButRegularFiles(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + commit := strings.Repeat("a", 40) + knot.setTree(repo, commit, + regular("ci.yaml"), + defsFile{"exec.yaml", "0100755", defsWorkflow}, + defsFile{"link.yaml", "0120000", "/etc/passwd"}, + defsFile{"sub", "0160000", ""}, + defsFile{"nested", "0040000", ""}, + ) + t.Setenv("PATH", "") + + names, _ := definitionNames(t, s, repo, commit) + if strings.Join(names, ",") != "ci.yaml,exec.yaml" { + t.Fatalf("definitions = %v, want only the regular files", names) + } + if got := knot.count("org.tangled.git.getBlob"); got != 2 { + t.Fatalf("%d blob reads, want one per regular file", got) + } +} + +func TestDefinitionKnotReadRefusesMoreFilesThanSpindleRuns(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + t.Setenv("PATH", "") + + var files []defsFile + for i := range maxWorkflowFiles + 1 { + files = append(files, regular(fmt.Sprintf("w%03d.yaml", i))) + } + over := strings.Repeat("a", 40) + knot.setTree(repo, over, files...) + if _, err := s.ListWorkflowDefinitions(context.Background(), repo, over); !errors.Is(err, errTooManyWorkflowFiles) { + t.Fatalf("%d files read with err = %v, want errTooManyWorkflowFiles", len(files), err) + } + if got := knot.count("org.tangled.git.getBlob"); got > maxWorkflowFiles { + t.Fatalf("%d blobs read for a repo that gets refused", got) + } + + exact := strings.Repeat("c", 40) + knot.setTree(repo, exact, files[:maxWorkflowFiles]...) + if names, _ := definitionNames(t, s, repo, exact); len(names) != maxWorkflowFiles { + t.Fatalf("%d definitions, want %d", len(names), maxWorkflowFiles) + } +} + +func TestDefinitionKnotReadOfAMissingDirIsEmpty(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + t.Setenv("PATH", "") + + names, _ := definitionNames(t, s, repo, strings.Repeat("a", 40)) + if len(names) != 0 { + t.Fatalf("definitions = %v, want none", names) + } +} + +func TestDefinitionBranchRefsResolveToACommitBeforeReading(t *testing.T) { + for _, ref := range []string{"HEAD", "", "main", "refs/heads/main"} { + t.Run(ref, func(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + commit := strings.Repeat("d", 40) + knot.setBranch(repo, "main", commit) + knot.setTree(repo, commit, regular("ci.yaml")) + t.Setenv("PATH", "") + + names, commits := definitionNames(t, s, repo, ref) + if len(names) != 1 || commits[0] != commit { + t.Fatalf("definitions = %v at %v, want ci.yaml at %s", names, commits, commit) + } + for _, r := range knot.asked() { + if strings.HasPrefix(r, "org.tangled.git.getTree") || strings.HasPrefix(r, "org.tangled.git.getBlob") { + if !strings.Contains(r, " "+commit+" ") { + t.Fatalf("%q read by something other than the resolved commit", r) + } + } + } + }) + } +} + +func TestDefinitionBranchRefReadsLocallyAfterResolving(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + _, commit := cloneWithCommit(t, s, repo) + knot.setBranch(repo, "main", commit) + + _, commits := definitionNames(t, s, repo, "main") + if commits[0] != commit { + t.Fatalf("commit = %s, want %s", commits[0], commit) + } + if knot.count("org.tangled.git.getTree") != 0 || knot.count("org.tangled.git.getBlob") != 0 { + t.Fatalf("knot saw %v, want only the branch lookup", knot.asked()) + } +} + +func TestDescribeReadsTheSourceRepoNotTheTarget(t *testing.T) { + knot := newDefsKnot(t) + target, source := syntax.DID("did:plc:defstarget"), syntax.DID("did:plc:defssource") + s := newDefsSpindle(t, knot, target, source) + commit := strings.Repeat("a", 40) + knot.setTree(source, commit, regular("fork.yaml")) + t.Setenv("PATH", "") + + out, err := s.DescribeWorkflowDefinition(context.Background(), target, commit, source) + if err != nil { + t.Fatal(err) + } + if len(out.Workflows) != 1 || out.Workflows[0] != "fork.yaml" { + t.Fatalf("workflows = %v, want the fork's", out.Workflows) + } +} + +func TestDefinitionTagRefsResolveToTheirCommitWithoutAClone(t *testing.T) { + for _, ref := range []string{"v1.0", "refs/tags/v1.0"} { + t.Run(ref, func(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + commit := strings.Repeat("e", 40) + knot.setTag(repo, "v1.0", commit) + knot.setTree(repo, commit, regular("ci.yaml")) + t.Setenv("PATH", "") + + names, commits := definitionNames(t, s, repo, ref) + if len(names) != 1 || commits[0] != commit { + t.Fatalf("definitions = %v at %v, want ci.yaml at %s", names, commits, commit) + } + }) + } +} + +func TestDefinitionBareRefIsABranchBeforeATag(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + branchCommit, tagCommit := strings.Repeat("1", 40), strings.Repeat("2", 40) + knot.setBranch(repo, "release", branchCommit) + knot.setTag(repo, "release", tagCommit) + knot.setTree(repo, branchCommit, regular("ci.yaml")) + knot.setTree(repo, tagCommit, regular("old.yaml")) + t.Setenv("PATH", "") + + if names, _ := definitionNames(t, s, repo, "release"); len(names) != 1 || names[0] != "ci.yaml" { + t.Fatalf("definitions = %v, want the branch's", names) + } + if got := knot.count("org.tangled.git.getTag"); got != 0 { + t.Fatalf("%d tag lookups for a ref that is a branch", got) + } +} + +func TestDefinitionRefThatIsNeitherBranchNorTagFails(t *testing.T) { + knot := newDefsKnot(t) + repo := syntax.DID("did:plc:defsrepo") + s := newDefsSpindle(t, knot, repo) + t.Setenv("PATH", "") + if _, err := s.ListWorkflowDefinitions(context.Background(), repo, "ghost"); err == nil { + t.Fatal("an unknown ref resolved") + } +}