Show server-side progress during Git pushes

Makefile

@@ -25,7 +25,7 @@ gc-acceptance:
 
 GIT_EXAMPLE_DIR = examples/git
 WAL_COMPONENT = examples/wal-component/Cargo.toml
-GIT_TEST_FILES = access.go access_test.go auth.go auth_test.go repositories.go repositories_test.go imports.go imports_test.go receive_transport.go receive_transport_test.go request_config.go request_config_test.go import_pack.go import_pack_test.go import_pack_lifecycle_test.go limits.go limits_test.go index.go index_test.go codec.go codec_test.go delta.go delta_test.go read_retry.go read_retry_test.go retention.go retention_test.go retry.go retry_test.go fetch_policy.go fetch_policy_test.go validate_objects.go validate_objects_test.go validate.go validate_refname.go validate_refname_test.go validate_native_test.go
+GIT_TEST_FILES = access.go access_test.go auth.go auth_test.go repositories.go repositories_test.go imports.go imports_test.go receive_transport.go receive_transport_test.go receive_progress.go receive_progress_test.go request_config.go request_config_test.go import_pack.go import_pack_test.go import_pack_lifecycle_test.go limits.go limits_test.go index.go index_test.go codec.go codec_test.go delta.go delta_test.go read_retry.go read_retry_test.go retention.go retention_test.go retry.go retry_test.go fetch_policy.go fetch_policy_test.go validate_objects.go validate_objects_test.go validate.go validate_refname.go validate_refname_test.go validate_native_test.go
 
 git-check: git-qualification-tools-test
 	test -z "$$(gofmt -l $(GIT_EXAMPLE_DIR)/*.go $(GIT_EXAMPLE_DIR)/tests/*.go)"

examples/git/README.md

@@ -66,6 +66,12 @@ git clone http://127.0.0.1:19100/team/demo.git cloned-demo
 Git defaults to SHA-1. To use SHA-256, initialize the local repository with
 `git init --object-format=sha256`; its first push selects the remote format.
 
+Push progress shows pack checks, imported-object counts, delta indexing,
+validation, and publication. Import counts update about once a second as objects
+finish; final push status follows durable publication. Use `git push --progress`
+to request feedback when stderr is redirected. `-q` and `--no-progress` suppress
+these messages.
+
 Press Ctrl-C in the service terminal to stop Spin and MinIO and remove all demo
 data. Set `OBJECT_LOG_GIT_LOCAL_PORT` to choose another loopback port.
 To keep data or use an existing backend, build with `make git-build`, configure

examples/git/main.go

@@ -45,7 +45,10 @@ func advertise(w io.Writer, s *store, unknownFormat bool) error {
 		return err
 	}
 	adv := &packp.AdvRefs{}
-	for _, feature := range []string{capability.ReportStatus, capability.DeleteRefs, capability.OFSDelta, capability.Atomic, capability.NoThin} {
+	for _, feature := range []string{
+		capability.ReportStatus, capability.DeleteRefs, capability.OFSDelta,
+		capability.Atomic, capability.NoThin, capability.Sideband64k, capability.Quiet,
+	} {
 		adv.Capabilities.Add(feature)
 	}
 	adv.Capabilities.Set(capability.ObjectFormat, s.meta.Format.String())
@@ -166,8 +169,9 @@ func serve(response http.ResponseWriter, r *http.Request) {
 		http.Error(response, err.Error(), operationStatus(err))
 		return
 	}
+	var pushCapabilities *capability.List
 	if service == transport.ReceivePackService && method == http.MethodPost {
-		format, body, err := receiveFormat(r.Body, limits.negotiationBytes)
+		format, capabilities, body, err := receiveFormat(r.Body, limits.negotiationBytes)
 		if err == nil && format != "" && route.Repository.Format != "" && route.Repository.Format != format {
 			err = config.ErrInvalidObjectFormat
 		}
@@ -186,6 +190,7 @@ func serve(response http.ResponseWriter, r *http.Request) {
 			return
 		}
 		route.Repository.Format = format
+		pushCapabilities = capabilities
 		r.Body = gitio.NewReadCloser(body, r.Body)
 	}
 
@@ -352,12 +357,19 @@ func serve(response http.ResponseWriter, r *http.Request) {
 		e = advertise(w, s, s.stateRoot == nil && route.Repository.Format == "")
 	} else if service == transport.ReceivePackService {
 		w.Header().Set("Content-Type", "application/x-git-receive-pack-result")
+		s.progress = newReceiveProgress(w, pushCapabilities, cancel)
 		commands := &commandReader{LimitedReader: io.LimitedReader{R: r.Body, N: limits.negotiationBytes}}
 		push := &receiveStore{Storer: s, commands: commands}
 		e = transport.ReceivePack(r.Context(), push, gitio.NewReadCloser(commands, r.Body), gitio.WriteNopCloser(w), &transport.ReceivePackRequest{StatelessRPC: true, Hooks: transport.ReceivePackHooks{PreReceive: func(_ context.Context, info *transport.PreReceiveInfo) error {
 			if len(info.Commands) == 0 {
 				return nil
 			}
+			if s.progress != nil {
+				s.progress.writer = info.Progress
+			}
+			if err := s.progress.message("Validating update...\n"); err != nil {
+				return err
+			}
 			refs, e := validate(s, info.Commands)
 			if e != nil {
 				return e
@@ -370,6 +382,9 @@ func serve(response http.ResponseWriter, r *http.Request) {
 					}
 				}
 			}
+			if err := s.progress.message("Publishing update...\n"); err != nil {
+				return err
+			}
 			e = s.publish(refs)
 			return e
 		}}})

examples/git/receive.go

@@ -46,6 +46,9 @@ func (p *incomingPack) Close() (err error) {
 	if p.err != nil {
 		return p.err
 	}
+	if err := p.s.progress.message("Checking received pack...\n"); err != nil {
+		return err
+	}
 	root, err := p.finish()
 	if err != nil {
 		return err
@@ -58,7 +61,10 @@ func (p *incomingPack) Close() (err error) {
 	}
 	defer reader.Close()
 	offsets := packOffsets{}
-	if err := importPack(p.s.ctx, reader, p.s, p.s.meta.Format, p.s.limits, offsets); err != nil {
+	if err := importPack(p.s.ctx, reader, p.s, p.s.meta.Format, p.s.limits, offsets, p.s.progress); err != nil {
+		return err
+	}
+	if err := p.s.progress.message("Indexing deltas...\n"); err != nil {
 		return err
 	}
 	remaining := p.s.limits.catalogBytes

examples/git/receive_progress.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/go-git/go-git/v6/plumbing"
+	"github.com/go-git/go-git/v6/plumbing/protocol/capability"
+	"github.com/go-git/go-git/v6/plumbing/protocol/packp/sideband"
+)
+
+// ReceivePack creates its hook progress writer after pack import. Use the same
+// upstream mux for import, then its hook writer for validation and publication.
+type receiveProgress struct {
+	writer      io.Writer
+	cancel      context.CancelFunc
+	total, done uint32
+	last        time.Time
+	err         error
+}
+
+func newReceiveProgress(w io.Writer, caps *capability.List, cancel context.CancelFunc) *receiveProgress {
+	if caps == nil || !caps.Supports(capability.Sideband64k) {
+		return nil
+	}
+	if caps.Supports(capability.Quiet) || caps.Supports(capability.NoProgress) {
+		return nil
+	}
+	return &receiveProgress{
+		writer: progressChannel{mux: sideband.NewMuxer(sideband.Sideband64k, w)},
+		cancel: cancel,
+	}
+}
+
+type progressChannel struct{ mux *sideband.Muxer }
+
+func (w progressChannel) Write(data []byte) (int, error) {
+	return w.mux.WriteChannel(sideband.ProgressMessage, data)
+}
+
+func (p *receiveProgress) message(text string) error {
+	if p == nil {
+		return nil
+	}
+	if p.err == nil {
+		_, p.err = io.WriteString(p.writer, text)
+		if p.err != nil && p.cancel != nil {
+			p.cancel()
+		}
+		p.last = time.Now()
+	}
+	return p.err
+}
+
+func (p *receiveProgress) OnHeader(total uint32) error {
+	if p == nil {
+		return nil
+	}
+	p.total = total
+	return p.message(fmt.Sprintf("Importing objects: 0/%d\n", total))
+}
+
+func (p *receiveProgress) OnInflatedObjectHeader(plumbing.ObjectType, int64, int64) error {
+	if p == nil {
+		return nil
+	}
+	return p.err
+}
+
+func (p *receiveProgress) OnInflatedObjectContent(plumbing.Hash, int64, uint32, []byte) error {
+	if p == nil {
+		return nil
+	}
+	p.done++
+	if time.Since(p.last) < time.Second {
+		return p.err
+	}
+	return p.message(fmt.Sprintf("Importing objects: %d/%d\r", p.done, p.total))
+}
+
+func (p *receiveProgress) OnFooter(plumbing.Hash) error {
+	if p == nil {
+		return nil
+	}
+	return p.message(fmt.Sprintf("Importing objects: %d/%d\n", p.done, p.total))
+}