diff --git a/appview/repo/index.go b/appview/repo/index.go index f6782c26..6a2e4d6b 100644 --- a/appview/repo/index.go +++ b/appview/repo/index.go @@ -116,7 +116,9 @@ func (rp *Repo) Index(w http.ResponseWriter, r *http.Request) { var languageInfo []types.RepoLanguageDetails if !result.IsEmpty { // TODO: a bit dirty - languageInfo, err = rp.getLanguageInfo(r.Context(), l, f, result.Ref, ref == "") + langCtx, cancel := context.WithTimeout(r.Context(), 1*time.Second) + defer cancel() + languageInfo, err = rp.getLanguageInfo(langCtx, l, f, result.Ref, ref == "") if err != nil { l.Warn("failed to compute language percentages", "err", err) // non-fatal diff --git a/go.mod b/go.mod index bac21511..5a442303 100644 --- a/go.mod +++ b/go.mod @@ -24,6 +24,8 @@ require ( github.com/cyphar/filepath-securejoin v0.4.1 github.com/dgraph-io/ristretto v0.2.0 github.com/did-method-plc/go-didplc v0.2.2 + github.com/djherbis/buffer v1.2.0 + github.com/djherbis/nio/v3 v3.0.1 github.com/docker/docker v28.2.2+incompatible github.com/dustin/go-humanize v1.0.1 github.com/gliderlabs/ssh v0.3.8 diff --git a/go.sum b/go.sum index 07e60c45..fcad8535 100644 --- a/go.sum +++ b/go.sum @@ -193,6 +193,11 @@ github.com/did-method-plc/go-didplc v0.2.2 h1:53HFhTT8NCAeFmZ6fdIZCf3PGDvj7A3cDj github.com/did-method-plc/go-didplc v0.2.2/go.mod h1:bKdJ21irnwNHgVLWWL32zUWqZueXYbJRUcxplZghByo= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/djherbis/buffer v1.1.0/go.mod h1:VwN8VdFkMY0DCALdY8o00d3IZ6Amz/UNVMWcSaJT44o= +github.com/djherbis/buffer v1.2.0 h1:PH5Dd2ss0C7CRRhQCZ2u7MssF+No9ide8Ye71nPHcrQ= +github.com/djherbis/buffer v1.2.0/go.mod h1:fjnebbZjCUpPinBRD+TDwXSOeNQ7fPQWLfGQqiAiUyE= +github.com/djherbis/nio/v3 v3.0.1 h1:6wxhnuppteMa6RHA4L81Dq7ThkZH8SwnDzXDYy95vB4= +github.com/djherbis/nio/v3 v3.0.1/go.mod h1:Ng4h80pbZFMla1yKzm61cF0tqqilXZYrogmWgZxOcmg= github.com/dlclark/regexp2 v1.7.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/dlclark/regexp2 v1.11.5 h1:Q/sSnsKerHeCkc/jSTNq1oCm7KiVgUMZRDUoRu0JQZQ= github.com/dlclark/regexp2 v1.11.5/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= diff --git a/knotmirror/xrpc/git_get_blob.go b/knotmirror/xrpc/git_get_blob.go index 66bd8ac0..8f5fea76 100644 --- a/knotmirror/xrpc/git_get_blob.go +++ b/knotmirror/xrpc/git_get_blob.go @@ -7,14 +7,13 @@ import ( "io" "net/http" "path/filepath" - "runtime/pprof" "slices" "strings" "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-git/go-git/v5/plumbing/object" - "tangled.org/core/knotserver/git" + "tangled.org/core/knotmirror/xrpc/gitea" ) func (x *Xrpc) GetBlob(w http.ResponseWriter, r *http.Request) { @@ -30,7 +29,7 @@ func (x *Xrpc) GetBlob(w http.ResponseWriter, r *http.Request) { return } - l := x.logger.With("repo", repo, "ref", ref, "path", path) + l := x.logger.With("method", "git.getBlob", "repo", repo, "ref", ref, "path", path) l.Debug("request") if path == "" { @@ -38,11 +37,8 @@ func (x *Xrpc) GetBlob(w http.ResponseWriter, r *http.Request) { return } - var file *object.File - pprof.Do(r.Context(), pprof.Labels("repo", repo.String()), func(ctx context.Context) { - file, err = x.getFile(ctx, repo, ref, path) - }) - if err != nil || file.Size > 1000*1000 { + size, reader, err := x.getFile(r.Context(), repo, ref, path) + if err != nil { l.Warn("local mirror failed, trying proxy", "err", err) if x.proxyToKnot(w, r, repo) { return @@ -50,17 +46,10 @@ func (x *Xrpc) GetBlob(w http.ResponseWriter, r *http.Request) { writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalServerError", Message: "failed to get blob"}) return } - - reader, err := file.Reader() - if err != nil { - l.Error("failed to read blob", "err", err) - writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalServerError", Message: "failed to read the blob"}) - return - } defer reader.Close() // default to octet-stream for large blobs - if file.Size > 1000*1000 { // 1MB + if size > 1000*1000 { // 1MB w.Header().Set("Content-Type", "application/octet-stream") if _, err := io.Copy(w, reader); err != nil { l.Error("failed to serve the blob", "err", err) @@ -111,18 +100,52 @@ func (x *Xrpc) GetBlob(w http.ResponseWriter, r *http.Request) { w.Write(contents) } -func (x *Xrpc) getFile(ctx context.Context, repo syntax.DID, ref, path string) (*object.File, error) { +func (x *Xrpc) getFile(ctx context.Context, repo syntax.DID, ref, path string) (int64, io.ReadCloser, error) { repoPath, err := x.makeRepoPath(ctx, repo) if err != nil { - return nil, fmt.Errorf("resolving repo did: %w", err) + return 0, nil, fmt.Errorf("resolving repo did: %w", err) + } + + rev := ref + if rev == "" { + rev = "HEAD" + } + + head, err := gitea.GetCommit(ctx, repoPath, rev) + if err != nil { + return 0, nil, fmt.Errorf("get head commit: %w", err) } - gr, err := git.Open(repoPath, ref) + treePath := filepath.Dir(path) + name := filepath.Base(path) + + // find subTree + subRev := head.Hash.String() + "^{tree}" + if treePath != "." { + subRev = head.Hash.String() + ":" + treePath + } + subTree, err := gitea.GetTree(ctx, repoPath, subRev) if err != nil { - return nil, fmt.Errorf("opening git repo: %w", err) + return 0, nil, fmt.Errorf("get subtree %s: %w", subRev, err) } - return gr.File(path) + // find entry + entry, err := func(subTree *object.Tree) (*object.TreeEntry, error) { + for _, entry := range subTree.Entries { + if entry.Name == name { + return &entry, nil + } + } + return nil, fmt.Errorf("object doesn't exist") + }(subTree) + if err != nil { + return 0, nil, fmt.Errorf("get file: %w", err) + } + + x.logger.Debug("ReadBlob", "name", entry.Name, "mode", entry.Mode.String(), "hash", entry.Hash.String()) + + // find blob + return gitea.ReadBlob(ctx, repoPath, entry.Hash) } var textualMimeTypes = []string{ diff --git a/knotmirror/xrpc/git_get_tree.go b/knotmirror/xrpc/git_get_tree.go index 3897976f..17b10683 100644 --- a/knotmirror/xrpc/git_get_tree.go +++ b/knotmirror/xrpc/git_get_tree.go @@ -3,18 +3,23 @@ package xrpc import ( "context" "fmt" + "io" "net/http" "path/filepath" - "runtime/pprof" "time" - "unicode/utf8" "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" + + "github.com/go-git/go-git/v5/plumbing" + "github.com/go-git/go-git/v5/plumbing/object" "tangled.org/core/api/tangled" - "tangled.org/core/appview/pages/markup" - "tangled.org/core/knotserver/git" - "tangled.org/core/types" + "tangled.org/core/knotmirror/xrpc/gitea" +) + +const ( + LastCommitCache = "last_commit:%s:%s" + LastCommitCacheTTL = 30 * 24 * time.Hour ) func (x *Xrpc) GetTree(w http.ResponseWriter, r *http.Request) { @@ -33,9 +38,7 @@ func (x *Xrpc) GetTree(w http.ResponseWriter, r *http.Request) { } var out *tangled.GitTempGetTree_Output - pprof.Do(r.Context(), pprof.Labels("repo", repo.String()), func(ctx context.Context) { - out, err = x.getTree(ctx, repo, ref, path) - }) + out, err = x.getTree(r.Context(), repo, ref, path) if err != nil { l.Warn("local mirror failed, trying proxy", "repo", repo, "err", err) if x.proxyToKnot(w, r, repo) { @@ -47,110 +50,171 @@ func (x *Xrpc) GetTree(w http.ResponseWriter, r *http.Request) { writeJson(w, http.StatusOK, out) } -func (x *Xrpc) getTree(ctx context.Context, repo syntax.DID, ref, path string) (*tangled.GitTempGetTree_Output, error) { +func (x *Xrpc) getTree(ctx context.Context, repo syntax.DID, ref, treePath string) (*tangled.GitTempGetTree_Output, error) { repoPath, err := x.makeRepoPath(ctx, repo) if err != nil { return nil, fmt.Errorf("failed to resolve repo did: %w", err) } + rev := ref + if rev == "" { + rev = "HEAD" + } - gr, err := git.Open(repoPath, ref) + head, err := gitea.GetCommit(ctx, repoPath, rev) if err != nil { - return nil, fmt.Errorf("opening git repo: %w", err) + return nil, fmt.Errorf("get head commit: %w", err) } - files, err := gr.FileTree(ctx, path) + subRev := head.Hash.String() + "^{tree}" + if treePath != "" { + subRev = head.Hash.String() + ":" + treePath + } + subTree, err := gitea.GetTree(ctx, repoPath, subRev) if err != nil { - return nil, fmt.Errorf("reading file tree: %w", err) + return nil, fmt.Errorf("get subtree %s: %w", subRev, err) } - // if any of these files are a readme candidate, pass along its blob contents too - var readmeFileName string - var readmeContents string - for _, file := range files { - if markup.IsReadmeFile(file.Name) { - contents, err := gr.RawContent(filepath.Join(path, file.Name)) - if err != nil { - x.logger.Error("failed to read contents of file", "path", path, "file", file.Name) - } + entryPaths := make([]string, len(subTree.Entries)+1) + entryPaths[0] = "" + for i, entry := range subTree.Entries { + entryPaths[i+1] = entry.Name + } + + commits, lastCommit, err := func(ctx context.Context, commit *object.Commit, treePath string, paths []string) (map[string]*object.Commit, *object.Commit, error) { + headRef := commit.Hash.String() + + revs := make(map[string]string, len(paths)) + var unHitPaths []string - if utf8.Valid(contents) { - readmeFileName = file.Name - readmeContents = string(contents) - break + keys := make([]string, len(paths)) + for i, path := range paths { + keys[i] = fmt.Sprintf(LastCommitCache, headRef, filepath.Join(treePath, path)) + } + if cached, err := x.rdb.MGet(ctx, keys...).Result(); err == nil { + for i, v := range cached { + if s, ok := v.(string); ok && s != "" { + revs[paths[i]] = s + } else { + unHitPaths = append(unHitPaths, paths[i]) + } } + } else { + unHitPaths = paths } - } - // convert NiceTree -> tangled.RepoTempGetTree_TreeEntry - treeEntries := make([]*tangled.GitTempGetTree_TreeEntry, len(files)) - for i, file := range files { - entry := &tangled.GitTempGetTree_TreeEntry{ - Name: file.Name, - Mode: file.Mode, - Size: file.Size, - } - if file.LastCommit != nil { - entry.Last_commit = &tangled.GitTempGetTree_LastCommit{ - Hash: file.LastCommit.Hash.String(), - Message: file.LastCommit.Message, - When: file.LastCommit.When.Format(time.RFC3339), + if len(unHitPaths) > 0 { + commits, err := gitea.WalkGitLog(ctx, repoPath, headRef, treePath, unHitPaths...) + if err != nil { + return nil, nil, err + } + pipe := x.rdb.Pipeline() + for path, cid := range commits { + if cid == "" { + continue + } + revs[path] = cid + pipe.Set(ctx, fmt.Sprintf(LastCommitCache, headRef, filepath.Join(treePath, path)), cid, LastCommitCacheTTL) + } + if _, err := pipe.Exec(ctx); err != nil { + x.logger.Warn("git last-commit cache write failed", "err", err) } } - treeEntries[i] = entry - } - var parentPtr *string - if path != "" { - parentPtr = &path - } + // start cat-file batch + batchWriter, batchReader, cancel := gitea.CatFileBatch(ctx, repoPath) + defer cancel() - var dotdotPtr *string - if path != "" { - dotdot := filepath.Dir(path) - if dotdot != "." { - dotdotPtr = &dotdot + // path -> commit map + commitsMap := map[string]*object.Commit{} + for path, commitId := range revs { + if commitId == headRef { + commitsMap[path] = commit + continue + } + + if commitId == "" { // invalid commit? + continue + } + + _, err := batchWriter.Write([]byte(commitId + "\n")) + if err != nil { + return nil, nil, err + } + _, typ, size, err := gitea.ReadBatchLine(batchReader) + if err != nil { + return nil, nil, err + } + if typ != "commit" { + if err := gitea.DiscardFull(batchReader, size+1); err != nil { + return nil, nil, err + } + return nil, nil, fmt.Errorf("unexpected type: %s for commit id: %s", typ, commitId) + } + c, err := gitea.ReadCommit(plumbing.NewHash(commitId), io.LimitReader(batchReader, size)) + if _, err := batchReader.Discard(1); err != nil { + return nil, nil, err + } + commitsMap[path] = c } - } - // find the most recent commit across all entries for the directory-level last commit - var lastCommitInfo *types.LastCommitInfo - for _, file := range files { - if file.LastCommit == nil { - continue + var treeCommit *object.Commit + if treePath == "" { + treeCommit = commit + } else if c, ok := commitsMap[""]; ok { + treeCommit = c } - if lastCommitInfo == nil { - lastCommitInfo = file.LastCommit - continue + + return commitsMap, treeCommit, nil + }(ctx, head, treePath, entryPaths) + if err != nil { + return nil, err + } + + outEntries := make([]*tangled.GitTempGetTree_TreeEntry, len(subTree.Entries)) + for i, entry := range subTree.Entries { + var entryLastCommit *tangled.GitTempGetTree_LastCommit + if commit, ok := commits[entry.Name]; ok { + entryLastCommit = &tangled.GitTempGetTree_LastCommit{ + Hash: commit.Hash.String(), + Message: commit.Message, + When: commit.Author.When.Format(time.RFC3339), + } } - if file.LastCommit.When.After(lastCommitInfo.When) { - lastCommitInfo = file.LastCommit + outEntries[i] = &tangled.GitTempGetTree_TreeEntry{ + Name: entry.Name, + Mode: entry.Mode.String(), + Last_commit: entryLastCommit, } } - var lastCommit *tangled.GitTempGetTree_LastCommit - if lastCommitInfo != nil { - lastCommit = &tangled.GitTempGetTree_LastCommit{ - Hash: lastCommitInfo.Hash.String(), - Message: lastCommitInfo.Message, - When: lastCommitInfo.When.Format(time.RFC3339), + var parent *string + var dotdot *string + if treePath != "" { + parent = &treePath + if dir := filepath.Dir(treePath); dir != "" { + dotdot = &dir } - if commit, err := gr.Commit(lastCommitInfo.Hash); err == nil { - lastCommit.Author = &tangled.GitTempGetTree_Signature{ - Name: commit.Author.Name, - Email: commit.Author.Email, - } + } + + var outLastCommit *tangled.GitTempGetTree_LastCommit + if lastCommit != nil { + outLastCommit = &tangled.GitTempGetTree_LastCommit{ + Hash: lastCommit.Hash.String(), + Message: lastCommit.Message, + When: lastCommit.Committer.When.Format(time.RFC3339), } } return &tangled.GitTempGetTree_Output{ Ref: ref, - Parent: parentPtr, - Dotdot: dotdotPtr, - Files: treeEntries, - LastCommit: lastCommit, + Parent: parent, + Dotdot: dotdot, + Files: outEntries, + LastCommit: outLastCommit, + // TODO: remove this field entirely Readme: &tangled.GitTempGetTree_Readme{ - Filename: readmeFileName, - Contents: readmeContents, + Filename: "", + Contents: "", }, }, nil } diff --git a/knotmirror/xrpc/git_list_languages.go b/knotmirror/xrpc/git_list_languages.go index fe1ce4e5..c63455e8 100644 --- a/knotmirror/xrpc/git_list_languages.go +++ b/knotmirror/xrpc/git_list_languages.go @@ -6,7 +6,6 @@ import ( "fmt" "math" "net/http" - "runtime/pprof" "time" "github.com/bluesky-social/indigo/atproto/atclient" @@ -49,9 +48,7 @@ func (x *Xrpc) ListLanguages(w http.ResponseWriter, r *http.Request) { } var out *tangled.GitTempListLanguages_Output - pprof.Do(r.Context(), pprof.Labels("repo", repo.String()), func(ctx context.Context) { - out, err = x.listLanguages(ctx, repo, ref) - }) + out, err = x.listLanguages(r.Context(), repo, ref) if err != nil { l.Warn("local mirror failed, trying proxy", "err", err) if x.proxyToKnot(w, r, repo) { @@ -61,6 +58,15 @@ func (x *Xrpc) ListLanguages(w http.ResponseWriter, r *http.Request) { return } + go func() { + ctx := context.Background() + encoded, err := json.Marshal(out.Languages) + if err != nil { + return + } + x.rdb.Set(ctx, fmt.Sprintf(RepoLanguagesByDid, repo, ref), encoded, RepoLanguagesTTL) + }() + writeJson(w, http.StatusOK, out) } @@ -83,17 +89,6 @@ func (x *Xrpc) listLanguages(ctx context.Context, repo syntax.DID, ref string) ( return nil, fmt.Errorf("analyzing languages: %w", err) } - langs := sizesToLanguages(sizes) - - go func() { - ctx := context.Background() - encoded, err := json.Marshal(langs) - if err != nil { - return - } - x.rdb.Set(ctx, fmt.Sprintf(RepoLanguagesByDid, repo, ref), encoded, RepoLanguagesTTL) - }() - return &tangled.GitTempListLanguages_Output{ Ref: ref, Languages: sizesToLanguages(sizes), diff --git a/knotmirror/xrpc/gitea/batch.go b/knotmirror/xrpc/gitea/batch.go new file mode 100644 index 00000000..017802f5 --- /dev/null +++ b/knotmirror/xrpc/gitea/batch.go @@ -0,0 +1,361 @@ +// NOTE: lot's of code compied from Gitea with slight modification to use go-git objects + +package gitea + +import ( + "bufio" + "bytes" + "context" + "fmt" + "io" + "math" + "os/exec" + "strconv" + "strings" + + "github.com/djherbis/buffer" + "github.com/djherbis/nio/v3" + "github.com/go-git/go-git/v5/plumbing" + "github.com/go-git/go-git/v5/plumbing/filemode" + "github.com/go-git/go-git/v5/plumbing/hash" + "github.com/go-git/go-git/v5/plumbing/object" +) + +func GetCommit(ctx context.Context, repoPath, rev string) (*object.Commit, error) { + wr, rd, cancel := CatFileBatch(ctx, repoPath) + defer cancel() + + if _, err := wr.Write([]byte(rev + "\n")); err != nil { + return nil, fmt.Errorf("write rev: %w", err) + } + sha, typ, size, err := ReadBatchLine(rd) + if err != nil { + return nil, err + } + if typ != "commit" { + if err := DiscardFull(rd, size+1); err != nil { + return nil, err + } + return nil, fmt.Errorf("unexpected type: %s for commit: %s", typ, rev) + } + commit, err := ReadCommit(plumbing.NewHash(string(sha)), io.LimitReader(rd, size)) + if err != nil { + return nil, fmt.Errorf("read commit %s: %w", rev, err) + } + if _, err := rd.Discard(1); err != nil { + return nil, err + } + return commit, nil +} + +func GetTree(ctx context.Context, repoPath, rev string) (*object.Tree, error) { + wr, rd, cancel := CatFileBatch(ctx, repoPath) + defer cancel() + + if _, err := wr.Write([]byte(rev + "\n")); err != nil { + return nil, fmt.Errorf("write rev: %w", err) + } + sha, typ, size, err := ReadBatchLine(rd) + if err != nil { + return nil, fmt.Errorf("resolve %s: %w", rev, err) + } + if typ != "tree" { + if err := DiscardFull(rd, size+1); err != nil { + return nil, err + } + return nil, fmt.Errorf("unexpected type: %s for tree: %s", typ, rev) + } + + entries, err := catBatchParseTreeEntries(rd, size) + if err != nil { + return nil, fmt.Errorf("read tree %s: %w", rev, err) + } + return &object.Tree{ + Hash: plumbing.NewHash(string(sha)), + Entries: entries, + }, nil +} + +func catBatchParseTreeEntries(rd *bufio.Reader, sz int64) ([]object.TreeEntry, error) { + entries := make([]object.TreeEntry, 0, 10) +loop: + for sz > 0 { + mode, fname, sha, count, err := ParseCatFileTreeLine(rd) + if err != nil { + if err == io.EOF { + break loop + } + return nil, err + } + modeNum, err := strconv.ParseUint(string(mode), 8, 32) + if err != nil { + return nil, err + } + sz -= int64(count) + entry := object.TreeEntry{ + Name: string(fname), + Mode: filemode.FileMode(modeNum), + Hash: plumbing.Hash(sha), + } + entries = append(entries, entry) + } + if _, err := rd.Discard(1); err != nil { + return entries, err + } + return entries, nil +} + +func CatFileBatch(ctx context.Context, repoPath string) (io.WriteCloser, *bufio.Reader, func()) { + batchStdinReader, batchStdinWriter := io.Pipe() + batchStdoutReader, batchStdoutWriter := nio.Pipe(buffer.New(32 * 1024)) + ctx, ctxCancel := context.WithCancel(ctx) + closed := make(chan struct{}) + cancel := func() { + ctxCancel() + _ = batchStdinWriter.Close() + _ = batchStdoutReader.Close() + <-closed + } + + // Ensure cancel is called as soon as the provided context is cancelled + go func() { + <-ctx.Done() + cancel() + }() + + go func() { + stderr := &strings.Builder{} + cmd := exec.CommandContext(ctx, "git", "-C", repoPath, "cat-file", "--batch") + cmd.Stdin = batchStdinReader + cmd.Stdout = batchStdoutWriter + cmd.Stderr = stderr + if err := cmd.Run(); err != nil { + _ = batchStdinReader.CloseWithError(fmt.Errorf("%w\n%s", err, stderr.String())) + _ = batchStdoutWriter.CloseWithError(fmt.Errorf("%w\n%s", err, stderr.String())) + } else { + _ = batchStdoutWriter.Close() + _ = batchStdinReader.Close() + } + close(closed) + }() + + batchReader := bufio.NewReaderSize(batchStdoutReader, 32*1024) + return batchStdinWriter, batchReader, cancel +} + +func ReadBatchLine(reader io.Reader) (sha []byte, typ string, size int64, err error) { + rd, ok := reader.(*bufio.Reader) + if !ok { + rd = bufio.NewReader(reader) + } + typ, err = rd.ReadString('\n') + if err != nil { + return sha, typ, size, err + } + if len(typ) == 1 { + typ, err = rd.ReadString('\n') + if err != nil { + return sha, typ, size, err + } + } + idx := strings.IndexByte(typ, ' ') + if idx < 0 { + return sha, typ, size, fmt.Errorf("missing sha: %s", sha) + } + sha = []byte(typ[:idx]) + typ = typ[idx+1:] + + idx = strings.IndexByte(typ, ' ') + if idx < 0 { + return sha, typ, size, fmt.Errorf("missing size: %s", sha) + } + + sizeStr := typ[idx+1 : len(typ)-1] + typ = typ[:idx] + + size, err = strconv.ParseInt(sizeStr, 10, 64) + return sha, typ, size, err +} + +// NOTE: readCommit doesn't return complete go-git [object.Commit] object! +// The embedded object store is missing, so calling method from returned commit +// can lead to panic. +func ReadCommit(oid plumbing.Hash, reader io.Reader) (*object.Commit, error) { + commit := &object.Commit{ + Hash: oid, + } + + payloadSB := new(strings.Builder) + signatureSB := new(strings.Builder) + messageSB := new(strings.Builder) + firstLine := true + message := false + pgpsig := false + + bufReader, ok := reader.(*bufio.Reader) + if !ok { + bufReader = bufio.NewReader(reader) + } + +readLoop: + for { + line, err := bufReader.ReadBytes('\n') + if err != nil { + if err == io.EOF { + if message { + _, _ = messageSB.Write(line) + } + _, _ = payloadSB.Write(line) + break readLoop + } + return nil, err + } + if pgpsig { + if len(line) > 0 && line[0] == ' ' { + _, _ = signatureSB.Write(line[1:]) + continue + } + pgpsig = false + } + + if !message { + // This is probably not correct but is copied from go-gits interpretation... + trimmed := bytes.TrimSpace(line) + if len(trimmed) == 0 { + message = true + _, _ = payloadSB.Write(line) + continue + } + + split := bytes.SplitN(trimmed, []byte{' '}, 2) + var data []byte + if len(split) > 1 { + data = split[1] + } + + switch string(split[0]) { + case "tree": + commit.TreeHash = plumbing.NewHash(string(data)) + _, _ = payloadSB.Write(line) + case "parent": + commit.ParentHashes = append(commit.ParentHashes, plumbing.NewHash(string(data))) + _, _ = payloadSB.Write(line) + case "author": + commit.Author.Decode(data) + _, _ = payloadSB.Write(line) + case "committer": + commit.Committer.Decode(data) + _, _ = payloadSB.Write(line) + case "gpgsig": + fallthrough + case "gpgsig-sha256": // FIXME: no intertop, so only 1 exists at present. + _, _ = signatureSB.Write(data) + _ = signatureSB.WriteByte('\n') + pgpsig = true + default: + // If the first line is not any of the known headers, then it is probably the prefix added when git cat-file is called with --batch, and that is not part of the payload + if !firstLine { + // Every subsequent header field is added to the payload + _, _ = payloadSB.Write(line) + } + } + } else { + _, _ = messageSB.Write(line) + _, _ = payloadSB.Write(line) + } + + firstLine = false + } + commit.Message = messageSB.String() + commit.PGPSignature = signatureSB.String() + + return commit, nil +} + +// ParseCatFileTreeLine reads an entry from a tree in a cat-file --batch stream +// This carefully avoids allocations - except where fnameBuf is too small. +// It is recommended therefore to pass in an fnameBuf large enough to avoid almost all allocations +// +// Each line is composed of: +// SP NUL +// +// We don't attempt to convert the raw HASH to save a lot of time +func ParseCatFileTreeLine(rd *bufio.Reader) (mode, fname, sha []byte, n int, err error) { + modeBuf := make([]byte, 40) + fnameBuf := make([]byte, 4096) + shaBuf := make([]byte, hash.HexSize) + + var readBytes []byte + + // Read the Mode & fname + readBytes, err = rd.ReadSlice('\x00') + if err != nil { + return mode, fname, sha, n, err + } + idx := bytes.IndexByte(readBytes, ' ') + if idx < 0 { + return mode, fname, sha, n, fmt.Errorf("missing") + } + + n += idx + 1 + copy(modeBuf, readBytes[:idx]) + if len(modeBuf) >= idx { + modeBuf = modeBuf[:idx] + } else { + modeBuf = append(modeBuf, readBytes[len(modeBuf):idx]...) + } + mode = modeBuf + + readBytes = readBytes[idx+1:] + + // Deal with the fname + copy(fnameBuf, readBytes) + if len(fnameBuf) > len(readBytes) { + fnameBuf = fnameBuf[:len(readBytes)] + } else { + fnameBuf = append(fnameBuf, readBytes[len(fnameBuf):]...) + } + for err == bufio.ErrBufferFull { + readBytes, err = rd.ReadSlice('\x00') + fnameBuf = append(fnameBuf, readBytes...) + } + n += len(fnameBuf) + if err != nil { + return mode, fname, sha, n, err + } + fnameBuf = fnameBuf[:len(fnameBuf)-1] + fname = fnameBuf + + // Deal with the binary hash + idx = 0 + length := hash.HexSize / 2 + for idx < length { + var read int + read, err = rd.Read(shaBuf[idx:length]) + n += read + if err != nil { + return mode, fname, sha, n, err + } + idx += read + } + sha = shaBuf + return mode, fname, sha, n, err +} + +func DiscardFull(rd *bufio.Reader, discard int64) error { + if discard > math.MaxInt32 { + n, err := rd.Discard(math.MaxInt32) + discard -= int64(n) + if err != nil { + return err + } + } + for discard > 0 { + n, err := rd.Discard(int(discard)) + discard -= int64(n) + if err != nil { + return err + } + } + return nil +} diff --git a/knotmirror/xrpc/gitea/blob.go b/knotmirror/xrpc/gitea/blob.go new file mode 100644 index 00000000..d87770e5 --- /dev/null +++ b/knotmirror/xrpc/gitea/blob.go @@ -0,0 +1,80 @@ +// Copyright 2021 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package gitea + +import ( + "bufio" + "bytes" + "context" + "io" + + "github.com/go-git/go-git/v5/plumbing" +) + +// ReadBlob returns blob size and [io.ReadCloser] of that blob. +func ReadBlob(ctx context.Context, repoPath string, hash plumbing.Hash) (int64, io.ReadCloser, error) { + wr, rd, cancel := CatFileBatch(ctx, repoPath) + + _, err := wr.Write([]byte(hash.String() + "\n")) + if err != nil { + cancel() + return 0, nil, err + } + _, _, size, err := ReadBatchLine(rd) + if err != nil { + cancel() + return 0, nil, err + } + + if size < 4096 { + bs, err := io.ReadAll(io.LimitReader(rd, size)) + defer cancel() + if err != nil { + return 0, nil, err + } + _, err = rd.Discard(1) + return size, io.NopCloser(bytes.NewReader(bs)), err + } + + return size, &blobReader{ + rd: rd, + n: size, + cancel: cancel, + }, nil +} + +type blobReader struct { + rd *bufio.Reader + n int64 + cancel func() +} + +func (b *blobReader) Read(p []byte) (n int, err error) { + if b.n <= 0 { + return 0, io.EOF + } + if int64(len(p)) > b.n { + p = p[0:b.n] + } + n, err = b.rd.Read(p) + b.n -= int64(n) + return n, err +} + +// Close implements io.Closer +func (b *blobReader) Close() error { + if b.rd == nil { + return nil + } + + defer b.cancel() + + if err := DiscardFull(b.rd, b.n+1); err != nil { + return err + } + + b.rd = nil + + return nil +} diff --git a/knotmirror/xrpc/gitea/gitea.go b/knotmirror/xrpc/gitea/gitea.go new file mode 100644 index 00000000..41697774 --- /dev/null +++ b/knotmirror/xrpc/gitea/gitea.go @@ -0,0 +1,410 @@ +// Copyright 2021 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package gitea + +import ( + "bufio" + "bytes" + "context" + "errors" + "fmt" + "io" + "os/exec" + "path" + "strings" + + "github.com/djherbis/buffer" + "github.com/djherbis/nio/v3" + "tangled.org/core/sets" +) + +// LogNameStatusRepo opens git log --raw in the provided repo and returns a stdin pipe, a stdout reader and cancel function +func LogNameStatusRepo(ctx context.Context, repository, headRef, treepath string, paths ...string) (*bufio.Reader, func()) { + // We often want to feed the commits in order into cat-file --batch, followed by their trees and sub trees as necessary. + // so let's create a batch stdin and stdout + stdoutReader, stdoutWriter := nio.Pipe(buffer.New(32 * 1024)) + + // Lets also create a context so that we can absolutely ensure that the command should die when we're done + ctx, ctxCancel := context.WithCancel(ctx) + + cancel := func() { + ctxCancel() + _ = stdoutReader.Close() + _ = stdoutWriter.Close() + } + + cmd := exec.CommandContext(ctx, + "git", + "log", + "--name-status", + "-c", + "--format=commit%x00%H %P%x00", + "--parents", + "--no-renames", + "-t", + "-z", + headRef, + ) + + var files []string + if len(paths) < 70 { + if treepath != "" { + files = append(files, treepath) + for _, pth := range paths { + if pth != "" { + files = append(files, path.Join(treepath, pth)) + } + } + } else { + for _, pth := range paths { + if pth != "" { + files = append(files, pth) + } + } + } + } else if treepath != "" { + files = append(files, treepath) + } + // Use the :(literal) pathspec magic to handle edge cases with files named like ":file.txt" or "*.jpg" + for i, file := range files { + files[i] = ":(literal)" + file + } + cmd.Args = append(cmd.Args, files...) + + go func() { + stderr := &strings.Builder{} + cmd.Dir = repository + cmd.Stdout = stdoutWriter + cmd.Stderr = stderr + if err := cmd.Run(); err != nil { + _ = stdoutWriter.CloseWithError(fmt.Errorf("%w\n%s", err, stderr.String())) + return + } + + _ = stdoutWriter.Close() + }() + + // For simplicities sake we'll us a buffered reader to read from the cat-file --batch + bufReader := bufio.NewReaderSize(stdoutReader, 32*1024) + + return bufReader, cancel +} + +// LogNameStatusRepoParser parses a git log raw output from LogRawRepo +type LogNameStatusRepoParser struct { + treepath string + paths []string + next []byte + buffull bool + rd *bufio.Reader + cancel func() +} + +// NewLogNameStatusRepoParser returns a new parser for a git log raw output +func NewLogNameStatusRepoParser(ctx context.Context, repository, head, treepath string, paths ...string) *LogNameStatusRepoParser { + rd, cancel := LogNameStatusRepo(ctx, repository, head, treepath, paths...) + return &LogNameStatusRepoParser{ + treepath: treepath, + paths: paths, + rd: rd, + cancel: cancel, + } +} + +// LogNameStatusCommitData represents a commit artefact from git log raw +type LogNameStatusCommitData struct { + CommitID string + ParentIDs []string + Paths []bool +} + +// Next returns the next LogStatusCommitData +func (g *LogNameStatusRepoParser) Next(treepath string, paths2ids map[string]int, changed []bool, maxpathlen int) (*LogNameStatusCommitData, error) { + var err error + if len(g.next) == 0 { + g.buffull = false + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + switch err { + case bufio.ErrBufferFull: + g.buffull = true + case io.EOF: + return nil, nil + default: + return nil, err + } + } + } + + ret := LogNameStatusCommitData{} + if bytes.Equal(g.next, []byte("commit\000")) { + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + switch err { + case bufio.ErrBufferFull: + g.buffull = true + case io.EOF: + return nil, nil + default: + return nil, err + } + } + } + + // Our "line" must look like: SP ( SP) * NUL + commitIDs := string(g.next) + if g.buffull { + more, err := g.rd.ReadString('\x00') + if err != nil { + return nil, err + } + commitIDs += more + } + commitIDs = commitIDs[:len(commitIDs)-1] + splitIDs := strings.Split(commitIDs, " ") + ret.CommitID = splitIDs[0] + if len(splitIDs) > 1 { + ret.ParentIDs = splitIDs[1:] + } + + // now read the next "line" + g.buffull = false + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + if err == bufio.ErrBufferFull { + g.buffull = true + } else if err != io.EOF { + return nil, err + } + } + + if err == io.EOF || (g.next[0] != '\n' && g.next[0] != '\000') { + return &ret, nil + } + + // Ok we have some changes. + // This line will look like: NL NUL + // + // Subsequent lines will not have the NL - so drop it here - g.bufffull must also be false at this point too. + if g.next[0] == '\n' { + g.next = g.next[1:] + } else { + g.buffull = false + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + if err == bufio.ErrBufferFull { + g.buffull = true + } else if err != io.EOF { + return nil, err + } + } + if len(g.next) == 0 { + return &ret, nil + } + if g.next[0] == '\x00' { + g.buffull = false + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + if err == bufio.ErrBufferFull { + g.buffull = true + } else if err != io.EOF { + return nil, err + } + } + } + } + + fnameBuf := make([]byte, 4096) + +diffloop: + for { + if err == io.EOF || bytes.Equal(g.next, []byte("commit\000")) { + return &ret, nil + } + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + switch err { + case bufio.ErrBufferFull: + g.buffull = true + case io.EOF: + return &ret, nil + default: + return nil, err + } + } + copy(fnameBuf, g.next) + if len(fnameBuf) < len(g.next) { + fnameBuf = append(fnameBuf, g.next[len(fnameBuf):]...) + } else { + fnameBuf = fnameBuf[:len(g.next)] + } + if err != nil { + if err != bufio.ErrBufferFull { + return nil, err + } + more, err := g.rd.ReadBytes('\x00') + if err != nil { + return nil, err + } + fnameBuf = append(fnameBuf, more...) + } + + // read the next line + g.buffull = false + g.next, err = g.rd.ReadSlice('\x00') + if err != nil { + if err == bufio.ErrBufferFull { + g.buffull = true + } else if err != io.EOF { + return nil, err + } + } + + if treepath != "" { + if !bytes.HasPrefix(fnameBuf, []byte(treepath)) { + fnameBuf = fnameBuf[:cap(fnameBuf)] + continue diffloop + } + } + fnameBuf = fnameBuf[len(treepath) : len(fnameBuf)-1] + if len(fnameBuf) > maxpathlen { + fnameBuf = fnameBuf[:cap(fnameBuf)] + continue diffloop + } + if len(fnameBuf) > 0 { + if len(treepath) > 0 { + if fnameBuf[0] != '/' || bytes.IndexByte(fnameBuf[1:], '/') >= 0 { + fnameBuf = fnameBuf[:cap(fnameBuf)] + continue diffloop + } + fnameBuf = fnameBuf[1:] + } else if bytes.IndexByte(fnameBuf, '/') >= 0 { + fnameBuf = fnameBuf[:cap(fnameBuf)] + continue diffloop + } + } + + idx, ok := paths2ids[string(fnameBuf)] + if !ok { + fnameBuf = fnameBuf[:cap(fnameBuf)] + continue diffloop + } + if ret.Paths == nil { + ret.Paths = changed + } + changed[idx] = true + } +} + +// Close closes the parser +func (g *LogNameStatusRepoParser) Close() { + g.cancel() +} + +// WalkGitLog walks the git log --name-status for the head commit in the provided treepath and files +func WalkGitLog(ctx context.Context, repoPath string, head, treepath string, paths ...string) (map[string]string, error) { + path2idx := map[string]int{} + maxpathlen := len(treepath) + + for i := range paths { + path2idx[paths[i]] = i + pthlen := len(paths[i]) + len(treepath) + 1 + if pthlen > maxpathlen { + maxpathlen = pthlen + } + } + + g := NewLogNameStatusRepoParser(ctx, repoPath, head, treepath, paths...) + // don't use defer g.Close() here as g may change its value - instead wrap in a func + defer func() { + g.Close() + }() + + results := make([]string, len(paths)) + remaining := len(paths) + nextRestart := min((len(paths)*3)/4, 70) + lastEmptyParent := head + commitSinceLastEmptyParent := uint64(0) + commitSinceNextRestart := uint64(0) + parentRemaining := sets.New[string]() + + changed := make([]bool, len(paths)) + +heaploop: + for { + select { + case <-ctx.Done(): + if ctx.Err() == context.DeadlineExceeded { + break heaploop + } + g.Close() + return nil, ctx.Err() + default: + } + current, err := g.Next(treepath, path2idx, changed, maxpathlen) + if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + break heaploop + } + g.Close() + return nil, err + } + if current == nil { + break heaploop + } + parentRemaining.Remove(current.CommitID) + for i, found := range current.Paths { + if !found { + continue + } + changed[i] = false + if results[i] == "" { + results[i] = current.CommitID + delete(path2idx, paths[i]) + remaining-- + if results[0] == "" { + results[0] = current.CommitID + delete(path2idx, "") + remaining-- + } + } + } + + if remaining <= 0 { + break heaploop + } + commitSinceLastEmptyParent++ + if parentRemaining.Len() == 0 { + lastEmptyParent = current.CommitID + commitSinceLastEmptyParent = 0 + } + if remaining <= nextRestart { + commitSinceNextRestart++ + if 4*commitSinceNextRestart > 3*commitSinceLastEmptyParent { + g.Close() + remainingPaths := make([]string, 0, len(paths)) + for i, pth := range paths { + if results[i] == "" { + remainingPaths = append(remainingPaths, pth) + } + } + g = NewLogNameStatusRepoParser(ctx, repoPath, lastEmptyParent, treepath, remainingPaths...) + parentRemaining = sets.New[string]() + nextRestart = (remaining * 3) / 4 + continue heaploop + } + } + for _, id := range current.ParentIDs { + parentRemaining.Insert(id) + } + } + g.Close() + + resultsMap := map[string]string{} + for i, pth := range paths { + resultsMap[pth] = results[i] + } + + return resultsMap, nil +} diff --git a/knotmirror/xrpc/repo_blob.go b/knotmirror/xrpc/repo_blob.go index 71d763b1..320e49ab 100644 --- a/knotmirror/xrpc/repo_blob.go +++ b/knotmirror/xrpc/repo_blob.go @@ -30,7 +30,7 @@ func (x *Xrpc) RepoBlob(w http.ResponseWriter, r *http.Request) { return } - l := x.logger.With("repo", repo, "ref", ref, "path", path) + l := x.logger.With("method", "repo.blob", "repo", repo, "ref", ref, "path", path) if path == "" { writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "missing path parameter"}) @@ -64,7 +64,7 @@ func (x *Xrpc) RepoBlob(w http.ResponseWriter, r *http.Request) { return } - file, err := x.getFile(r.Context(), repo, ref, path) + size, reader, err := x.getFile(r.Context(), repo, ref, path) if err != nil { l.Warn("local mirror failed, trying proxy", "err", err) if x.proxyToKnot(w, r, repo) { @@ -73,24 +73,19 @@ func (x *Xrpc) RepoBlob(w http.ResponseWriter, r *http.Request) { writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalServerError", Message: "failed to get blob"}) return } + defer reader.Close() - if file.Size > 1000*1000 { // 1MB + if size > 1000*1000 { // 1MB fileTooLarge := true writeJson(w, http.StatusOK, tangled.RepoBlob_Output{ Ref: ref, Path: path, - Size: &file.Size, + Size: &size, FileTooLarge: &fileTooLarge, }) return } - reader, err := file.Reader() - if err != nil { - l.Error("failed to read blob", "err", err) - writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalServerError", Message: "failed to read the blob"}) - return - } contents, err := io.ReadAll(reader) if err != nil { l.Error("failed to read blob content", "err", err) @@ -126,7 +121,7 @@ func (x *Xrpc) RepoBlob(w http.ResponseWriter, r *http.Request) { response := tangled.RepoBlob_Output{ Ref: ref, Path: path, - Size: &file.Size, + Size: &size, IsBinary: &isBinary, Content: content, } diff --git a/nix/gomod2nix.toml b/nix/gomod2nix.toml index d8396a46..595c2468 100644 --- a/nix/gomod2nix.toml +++ b/nix/gomod2nix.toml @@ -244,6 +244,12 @@ schema = 3 [mod."github.com/distribution/reference"] version = "v0.6.0" hash = "sha256-gr4tL+qz4jKyAtl8LINcxMSanztdt+pybj1T+2ulQv4=" + [mod."github.com/djherbis/buffer"] + version = "v1.2.0" + hash = "sha256-uHJQWcXwg4j2nOBK6epCm0gkv7KwriHZ1D7dgYqipp4=" + [mod."github.com/djherbis/nio/v3"] + version = "v3.0.1" + hash = "sha256-KZmVhlUht9phMpRtjiEMLjlyBwHKj5tnNQ2SC6pI7Rs=" [mod."github.com/dlclark/regexp2"] version = "v1.11.5" hash = "sha256-jN5+2ED+YbIoPIuyJ4Ou5pqJb2w1uNKzp5yTjKY6rEQ="