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 }