Add local batched Git blob reader
This commit is contained in:
222
internal/git/blob_reader.go
Normal file
222
internal/git/blob_reader.go
Normal file
@@ -0,0 +1,222 @@
|
||||
package git
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"gitea.maximumdirect.net/eric/ReleaseKit/internal/process"
|
||||
)
|
||||
|
||||
var errBlobObjectMissing = errors.New("Git object is missing")
|
||||
|
||||
// BlobReader reads requested blobs through one lazy git cat-file --batch
|
||||
// process. Reads are serialized; callers must close the reader when finished.
|
||||
type BlobReader struct {
|
||||
repositoryDir string
|
||||
parent context.Context
|
||||
|
||||
mu sync.Mutex
|
||||
child *process.PipedProcess
|
||||
stdout *bufio.Reader
|
||||
stderr bytes.Buffer
|
||||
closed bool
|
||||
aborted bool
|
||||
terminal error
|
||||
closeErr error
|
||||
}
|
||||
|
||||
// NewBlobReader creates a lazy reader for one repository. It starts no child
|
||||
// process until the first valid blob is requested.
|
||||
func NewBlobReader(ctx context.Context, repositoryDir string) *BlobReader {
|
||||
return &BlobReader{repositoryDir: repositoryDir, parent: ctx}
|
||||
}
|
||||
|
||||
// Read returns one blob body from the candidate repository. The returned bytes
|
||||
// are newly allocated and owned by the caller.
|
||||
func (reader *BlobReader) Read(ctx context.Context, objectID string) ([]byte, error) {
|
||||
if !validObjectID(objectID) {
|
||||
return nil, fmt.Errorf("invalid blob object ID %q", objectID)
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, fmt.Errorf("read blob %s: %w", objectID, err)
|
||||
}
|
||||
|
||||
reader.mu.Lock()
|
||||
defer reader.mu.Unlock()
|
||||
if reader.terminal != nil {
|
||||
return nil, reader.terminal
|
||||
}
|
||||
if reader.closed {
|
||||
return nil, fmt.Errorf("read blob %s: Git blob reader is closed", objectID)
|
||||
}
|
||||
if err := reader.startLocked(); err != nil {
|
||||
return nil, fmt.Errorf("read blob %s: %w", objectID, err)
|
||||
}
|
||||
|
||||
child := reader.child
|
||||
callbackDone := make(chan struct{})
|
||||
stopCancellation := context.AfterFunc(ctx, func() {
|
||||
defer close(callbackDone)
|
||||
child.Cancel()
|
||||
_ = child.Stdin().Close()
|
||||
_ = child.Stdout().Close()
|
||||
})
|
||||
body, err := reader.exchange(objectID)
|
||||
if !stopCancellation() {
|
||||
<-callbackDone
|
||||
}
|
||||
if contextErr := ctx.Err(); contextErr != nil {
|
||||
err = contextErr
|
||||
} else if contextErr := reader.parent.Err(); contextErr != nil {
|
||||
err = contextErr
|
||||
}
|
||||
if errors.Is(err, errBlobObjectMissing) {
|
||||
return nil, err
|
||||
}
|
||||
if err != nil {
|
||||
return nil, reader.abortLocked(objectID, err)
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
|
||||
// Close closes a healthy session cleanly or reaps an aborted session. It is
|
||||
// safe to call more than once.
|
||||
func (reader *BlobReader) Close() error {
|
||||
reader.mu.Lock()
|
||||
defer reader.mu.Unlock()
|
||||
if reader.closed {
|
||||
return reader.closeErr
|
||||
}
|
||||
reader.closed = true
|
||||
if reader.aborted {
|
||||
reader.closeErr = reader.terminal
|
||||
return reader.closeErr
|
||||
}
|
||||
if reader.child == nil {
|
||||
return nil
|
||||
}
|
||||
if err := reader.child.Stdin().Close(); err != nil {
|
||||
reader.closeErr = reader.abortLocked("", fmt.Errorf("close Git cat-file input: %w", err))
|
||||
return reader.closeErr
|
||||
}
|
||||
if _, err := io.Copy(io.Discard, reader.child.Stdout()); err != nil {
|
||||
reader.closeErr = reader.abortLocked("", fmt.Errorf("finish Git cat-file output: %w", err))
|
||||
return reader.closeErr
|
||||
}
|
||||
if err := reader.child.Wait(); err != nil {
|
||||
reader.closeErr = reader.processError("close Git cat-file batch", err)
|
||||
}
|
||||
return reader.closeErr
|
||||
}
|
||||
|
||||
func (reader *BlobReader) startLocked() error {
|
||||
if reader.child != nil {
|
||||
return nil
|
||||
}
|
||||
if err := reader.parent.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
child, err := process.StartPiped(reader.parent, process.Invocation{
|
||||
Args: []string{"git", "cat-file", "--batch"},
|
||||
Dir: reader.repositoryDir,
|
||||
Env: map[string]string{
|
||||
"GIT_NO_LAZY_FETCH": "1",
|
||||
"GIT_ALLOW_PROTOCOL": "",
|
||||
},
|
||||
Stderr: &reader.stderr,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
reader.child = child
|
||||
reader.stdout = bufio.NewReader(child.Stdout())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (reader *BlobReader) exchange(objectID string) ([]byte, error) {
|
||||
if _, err := io.WriteString(reader.child.Stdin(), objectID+"\n"); err != nil {
|
||||
return nil, fmt.Errorf("write object request: %w", err)
|
||||
}
|
||||
header, err := reader.stdout.ReadString('\n')
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read object header: %w", err)
|
||||
}
|
||||
fields := strings.Fields(strings.TrimSuffix(header, "\n"))
|
||||
if len(fields) == 2 && strings.EqualFold(fields[0], objectID) && fields[1] == "missing" {
|
||||
return nil, fmt.Errorf("read blob %s: %w", objectID, errBlobObjectMissing)
|
||||
}
|
||||
if len(fields) != 3 {
|
||||
return nil, fmt.Errorf("invalid Git blob header %q", strings.TrimSuffix(header, "\n"))
|
||||
}
|
||||
if !strings.EqualFold(fields[0], objectID) {
|
||||
return nil, fmt.Errorf("Git blob response object ID %q does not match request %q", fields[0], objectID)
|
||||
}
|
||||
if fields[1] != "blob" {
|
||||
return nil, fmt.Errorf("Git object %s has type %q, want blob", objectID, fields[1])
|
||||
}
|
||||
size, err := strconv.ParseInt(fields[2], 10, strconv.IntSize)
|
||||
if err != nil || size < 0 {
|
||||
if err == nil {
|
||||
err = errors.New("negative object size")
|
||||
}
|
||||
return nil, fmt.Errorf("invalid Git blob size %q: %w", fields[2], err)
|
||||
}
|
||||
body := make([]byte, int(size))
|
||||
if _, err := io.ReadFull(reader.stdout, body); err != nil {
|
||||
return nil, fmt.Errorf("read %d-byte Git blob body: %w", size, err)
|
||||
}
|
||||
framing, err := reader.stdout.ReadByte()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read Git blob framing: %w", err)
|
||||
}
|
||||
if framing != '\n' {
|
||||
return nil, fmt.Errorf("invalid Git blob framing byte %q", framing)
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
|
||||
func (reader *BlobReader) abortLocked(objectID string, cause error) error {
|
||||
if reader.aborted {
|
||||
return reader.terminal
|
||||
}
|
||||
reader.aborted = true
|
||||
var waitErr error
|
||||
if reader.child != nil {
|
||||
reader.child.Cancel()
|
||||
_ = reader.child.Stdin().Close()
|
||||
_ = reader.child.Stdout().Close()
|
||||
waitErr = reader.child.Wait()
|
||||
}
|
||||
if contextErr := reader.parent.Err(); contextErr != nil {
|
||||
cause = contextErr
|
||||
}
|
||||
reader.terminal = reader.processError("read blob "+objectID, errors.Join(cause, relevantProcessError(waitErr, cause)))
|
||||
return reader.terminal
|
||||
}
|
||||
|
||||
func (reader *BlobReader) processError(operation string, cause error) error {
|
||||
message := strings.TrimSpace(reader.stderr.String())
|
||||
if message != "" {
|
||||
return fmt.Errorf("%s: %s: %w", operation, message, cause)
|
||||
}
|
||||
return fmt.Errorf("%s: %w", operation, cause)
|
||||
}
|
||||
|
||||
func relevantProcessError(waitErr, cause error) error {
|
||||
if waitErr == nil || errors.Is(cause, context.Canceled) || errors.Is(cause, context.DeadlineExceeded) {
|
||||
return nil
|
||||
}
|
||||
var exitErr *exec.ExitError
|
||||
if errors.As(waitErr, &exitErr) && exitErr.ExitCode() == -1 {
|
||||
return nil
|
||||
}
|
||||
return waitErr
|
||||
}
|
||||
Reference in New Issue
Block a user