commands,tq: defer local verification to *tq.TransferQueue

Author: ttaylorr

commands/command_push.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/git-lfs/git-lfs/errors"
 	"github.com/git-lfs/git-lfs/git"
 	"github.com/git-lfs/git-lfs/lfs"
 	"github.com/rubyist/tracerx"
@@ -73,8 +74,22 @@ func uploadLeftOrAll(g *lfs.GitScanner, ctx *uploadContext, ref string) error {
 
 func uploadsWithObjectIDs(ctx *uploadContext, oids []string) {
 	for _, oid := range oids {
+		mp, err := lfs.LocalMediaPath(oid)
+		if err != nil {
+			ExitWithError(errors.Wrap(err, "Unable to find local media path:"))
+		}
+
+		stat, err := os.Stat(mp)
+		if err != nil {
+			ExitWithError(errors.Wrap(err, "Unable to stat local media path"))
+		}
+
 		uploadPointers(ctx, &lfs.WrappedPointer{
-			Pointer: &lfs.Pointer{Oid: oid},
+			Name: mp,
+			Pointer: &lfs.Pointer{
+				Oid:  oid,
+				Size: stat.Size(),
+			},
 		})
 	}
 

commands/commands.go

@@ -114,28 +114,26 @@ func downloadTransfer(p *lfs.WrappedPointer) (name, path, oid string, size int64
 	return p.Name, path, p.Oid, p.Size
 }
 
-func uploadTransfer(oid, filename string) (*tq.Transfer, error) {
+func uploadTransfer(p *lfs.WrappedPointer) (*tq.Transfer, error) {
+	filename := p.Name
+	oid := p.Oid
+
 	localMediaPath, err := lfs.LocalMediaPath(oid)
 	if err != nil {
 		return nil, errors.Wrapf(err, "Error uploading file %s (%s)", filename, oid)
 	}
 
 	if len(filename) > 0 {
-		if err = ensureFile(filename, localMediaPath); err != nil {
+		if err = ensureFile(filename, localMediaPath); err != nil && !errors.IsCleanPointerError(err) {
 			return nil, err
 		}
 	}
 
-	fi, err := os.Stat(localMediaPath)
-	if err != nil {
-		return nil, errors.Wrapf(err, "Error uploading file %s (%s)", filename, oid)
-	}
-
 	return &tq.Transfer{
 		Name: filename,
 		Path: localMediaPath,
 		Oid:  oid,
-		Size: fi.Size(),
+		Size: p.Size,
 	}, nil
 }
 
@@ -146,7 +144,6 @@ func ensureFile(smudgePath, cleanPath string) error {
 		return nil
 	}
 
-	expectedOid := filepath.Base(cleanPath)
 	localPath := filepath.Join(config.LocalWorkingDir, smudgePath)
 	file, err := os.Open(localPath)
 	if err != nil {
@@ -168,11 +165,6 @@ func ensureFile(smudgePath, cleanPath string) error {
 	if err != nil {
 		return err
 	}
-
-	if expectedOid != cleaned.Oid {
-		return fmt.Errorf("Trying to push %q with OID %s.\nNot found in %s.", smudgePath, expectedOid, filepath.Dir(cleanPath))
-	}
-
 	return nil
 }
 

commands/uploader.go

@@ -2,7 +2,6 @@ package commands
 
 import (
 	"os"
-	"sync"
 
 	"github.com/git-lfs/git-lfs/errors"
 	"github.com/git-lfs/git-lfs/lfs"
@@ -11,8 +10,6 @@ import (
 	"github.com/git-lfs/git-lfs/tq"
 )
 
-var uploadMissingErr = "%s does not exist in .git/lfs/objects. Tried %s, which matches %s."
-
 type uploadContext struct {
 	Remote       string
 	DryRun       bool
@@ -21,41 +18,20 @@ type uploadContext struct {
 
 	meter progress.Meter
 	tq    *tq.TransferQueue
-
-	cwg *sync.WaitGroup
-	cq  *tq.TransferQueue
 }
 
 func newUploadContext(remote string, dryRun bool) *uploadContext {
 	cfg.CurrentRemote = remote
 
-	return &uploadContext{
+	ctx := &uploadContext{
 		Remote:       remote,
 		Manifest:     getTransferManifest(),
 		DryRun:       dryRun,
 		uploadedOids: tools.NewStringSet(),
-
-		cwg: new(sync.WaitGroup),
-		// TODO(taylor): single item batches are needed to enqueue each
-		// item immediately to avoid waiting an infinite amount of time
-		// on an underfilled batch.
-		cq: newDownloadCheckQueue(c.Manifest, c.Remote, tq.WithBatchSize(1)),
 	}
 
-	ctx.cq.Notify(func(oid string, ok bool) {
-		if ok {
-			// If the object was "ok", the server already has it,
-			// and can be marked as uploaded.
-			ctx.SetUploaded(oid)
-		}
-
-		// No matter whether or not the sever has the object, mark this
-		// OID as checked.
-		ctx.cwg.Done()
-	})
-
 	ctx.meter = buildProgressMeter(ctx.DryRun)
-	ctx.tq = newUploadQueue(c.Manifest, c.Remote, tq.WithProgress(ctx.meter), tq.DryRun(ctx.DryRun))
+	ctx.tq = newUploadQueue(ctx.Manifest, ctx.Remote, tq.WithProgress(ctx.meter), tq.DryRun(ctx.DryRun))
 
 	return ctx
 }
@@ -75,8 +51,6 @@ func (c *uploadContext) HasUploaded(oid string) bool {
 func (c *uploadContext) prepareUpload(unfiltered ...*lfs.WrappedPointer) (*tq.TransferQueue, []*lfs.WrappedPointer) {
 	numUnfiltered := len(unfiltered)
 	uploadables := make([]*lfs.WrappedPointer, 0, numUnfiltered)
-	missingLocalObjects := make([]*lfs.WrappedPointer, 0, numUnfiltered)
-	missingSize := int64(0)
 
 	// XXX(taylor): temporary measure to fix duplicate (broken) results from
 	// scanner
@@ -96,47 +70,12 @@ func (c *uploadContext) prepareUpload(unfiltered ...*lfs.WrappedPointer) (*tq.Tr
 		// we will call Skip() based on the results of the download check queue.
 		c.meter.Add(p.Size)
 
-		if lfs.ObjectExistsOfSize(p.Oid, p.Size) {
-			uploadables = append(uploadables, p)
-		} else {
-			// We think we need to push this but we don't have it
-			// Store for server checking later
-			missingLocalObjects = append(missingLocalObjects, p)
-			missingSize += p.Size
-		}
-	}
-
-	// check to see if the server has the missing objects.
-	c.checkMissing(missingLocalObjects, missingSize)
-
-	// use the context's TransferQueue, automatically skipping any missing
-	// objects that the server already has.
-	for _, p := range missingLocalObjects {
-		if c.HasUploaded(p.Oid) {
-			// if the server already has this object, call Skip() on
-			// the progressmeter to decrement the number of files by
-			// 1 and the number of bytes by `p.Size`.
-			c.tq.Skip(p.Size)
-		} else {
-			uploadables = append(uploadables, p)
-		}
+		uploadables = append(uploadables, p)
 	}
 
 	return c.tq, uploadables
 }
 
-// This checks the given slice of pointers that don't exist in .git/lfs/objects
-// against the server. Anything the server already has does not need to be
-// uploaded again.
-func (c *uploadContext) checkMissing(missing []*lfs.WrappedPointer, missingSize int64) {
-	c.cwg.Add(len(missing))
-	for _, p := range missing {
-		c.cq.Add(downloadTransfer(p))
-	}
-
-	c.cwg.Wait()
-}
-
 func uploadPointers(c *uploadContext, unfiltered ...*lfs.WrappedPointer) {
 	if c.DryRun {
 		for _, p := range unfiltered {
@@ -153,13 +92,9 @@ func uploadPointers(c *uploadContext, unfiltered ...*lfs.WrappedPointer) {
 
 	q, pointers := c.prepareUpload(unfiltered...)
 	for _, p := range pointers {
-		t, err := uploadTransfer(p.Oid, p.Name)
-		if err != nil {
-			if errors.IsCleanPointerError(err) {
-				Exit(uploadMissingErr, p.Oid, p.Name, errors.GetContext(err, "pointer").(*lfs.Pointer).Oid)
-			} else {
-				ExitWithError(err)
-			}
+		t, err := uploadTransfer(p)
+		if err != nil && !errors.IsCleanPointerError(err) {
+			ExitWithError(err)
 		}
 
 		q.Add(t.Name, t.Path, t.Oid, t.Size)

test/test-pre-push.sh

@@ -189,7 +189,7 @@ begin_test "pre-push with missing pointer not on server"
     git lfs pre-push origin "$GITSERVER/$reponame" 2>&1 |
     tee push.log
   set -e
-  grep "7aa7a5359173d05b63cfd682e3c38487f3cb4f7f1d60659fe59fab1505977d4c does not exist in .git/lfs/objects. Tried new.dat, which matches 7aa7a5359173d05b63cfd682e3c38487f3cb4f7f1d60659fe59fab1505977d4c." push.log
+  grep "Unable to find object (7aa7a5359173d05b63cfd682e3c38487f3cb4f7f1d60659fe59fab1505977d4c) locally." push.log
 )
 end_test
 

tq/transfer_queue.go

@@ -1,6 +1,7 @@
 package tq
 
 import (
+	"os"
 	"sort"
 	"sync"
 
@@ -298,7 +299,6 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch batch) (batch, error)
 				next = append(next, t)
 			} else {
 				q.sendNotifications(t.Oid, false)
-				q.wait.Done()
 			}
 		}
 
@@ -404,16 +404,21 @@ func (q *TransferQueue) addToAdapter(e lfsapi.Endpoint, pending []*Transfer) <-c
 		return retries
 	}
 
+	present, missingResults := q.partitionTransfers(pending)
+
 	go func() {
 		defer close(retries)
 
 		var results <-chan TransferResult
 		if q.dryRun {
-			results = q.makeDryRunResults(pending)
+			results = q.makeDryRunResults(present)
 		} else {
-			results = q.adapter.Add(pending...)
+			results = q.adapter.Add(present...)
 		}
 
+		for _, res := range missingResults {
+			q.handleTransferResult(res, retries)
+		}
 		for res := range results {
 			q.handleTransferResult(res, retries)
 		}
@@ -422,6 +427,39 @@ func (q *TransferQueue) addToAdapter(e lfsapi.Endpoint, pending []*Transfer) <-c
 	return retries
 }
 
+func (q *TransferQueue) partitionTransfers(transfers []*Transfer) (present []*Transfer, results []TransferResult) {
+	if q.direction != Upload {
+		return transfers, nil
+	}
+
+	present = make([]*Transfer, 0, len(transfers))
+	results = make([]TransferResult, 0, len(transfers))
+
+	for _, t := range transfers {
+		var err error
+
+		if t.Size < 0 {
+			err = errors.Errorf("Git LFS: object %q has invalid size (got: %d)", t.Oid, t.Size)
+		} else {
+			fd, serr := os.Stat(t.Path)
+			if serr != nil || fd.Size() != t.Size {
+				err = errors.Errorf("Unable to find object (%s) locally.", t.Oid)
+			}
+		}
+
+		if err != nil {
+			results = append(results, TransferResult{
+				Transfer: t,
+				Error:    err,
+			})
+		} else {
+			present = append(present, t)
+		}
+	}
+
+	return
+}
+
 // makeDryRunResults returns a channel populated immediately with "successful"
 // results for all of the given transfers in "ts".
 func (q *TransferQueue) makeDryRunResults(ts []*Transfer) <-chan TransferResult {