tq: add exponential backoff for retries

Previously, retries happened as fast as possible unless the server
provided the Retry-After header. This is effective for certain types of
errors, but not when the LFS server is experiencing a temporary but not
instantaneous failure. Delaying between retries lets the server recover
and the LFS operation complete.

Delays start at a fixed 250ms for the first retry and double with each
successive retry up to a configurable maximum delay, 10s by default. The
maximum retry is configurable using lfs.transfer.maxretrydelay. Delays
can be disabled by setting the max delay to 0.

Author: bluekeyes

docs/man/git-lfs-config.5.ronn

@@ -160,6 +160,18 @@ be scoped inside the configuration for a remote.
   not an integer, is less than one, or is not given, a value of eight will be
   used instead.
 
+* `lfs.transfer.maxretrydelay`
+
+  Specifies the maximum time in seconds LFS will wait between each retry
+  attempt. LFS uses exponential backoff for retries, doubling the time between
+  each retry until reaching this limit. If a server requests a delay using the
+  `Retry-After` header, the header value overrides the exponential delay for
+  that attempt and is not limited by this option.
+  
+  Must be an integer which is not negative. Use zero to disable delays between
+  retries unless requested by a server. If the value is not an integer, is
+  negative, or is not given, a value of ten will be used instead.
+
 * `lfs.transfer.maxverifies`
 
   Specifies how many verification requests LFS will attempt per OID before

t/t-push.sh

@@ -575,7 +575,7 @@ begin_test "push (retry with expired actions)"
 
   GIT_TRACE=1 git push origin master 2>&1 | tee push.log
 
-  expected="enqueue retry #1 for \"$contents_oid\" (size: $contents_size): LFS: tq: action \"upload\" expires at"
+  expected="enqueue retry #1 after 0.25s for \"$contents_oid\" (size: $contents_size): LFS: tq: action \"upload\" expires at"
 
   grep "$expected" push.log
   grep "Uploading LFS objects: 100% (1/1), 21 B" push.log

tq/manifest.go

@@ -12,13 +12,16 @@ import (
 
 const (
 	defaultMaxRetries          = 8
+	defaultMaxRetryDelay       = 10
 	defaultConcurrentTransfers = 8
 )
 
 type Manifest struct {
 	// maxRetries is the maximum number of retries a single object can
-	// attempt to make before it will be dropped.
+	// attempt to make before it will be dropped. maxRetryDelay is the maximum
+	// time in seconds to wait between retry attempts when using backoff.
 	maxRetries              int
+	maxRetryDelay           int
 	concurrentTransfers     int
 	basicTransfersOnly      bool
 	standaloneTransferAgent string
@@ -39,6 +42,10 @@ func (m *Manifest) MaxRetries() int {
 	return m.maxRetries
 }
 
+func (m *Manifest) MaxRetryDelay() int {
+	return m.maxRetryDelay
+}
+
 func (m *Manifest) ConcurrentTransfers() int {
 	return m.concurrentTransfers
 }
@@ -77,6 +84,9 @@ func NewManifest(f *fs.Filesystem, apiClient *lfsapi.Client, operation, remote s
 		if v := git.Int("lfs.transfer.maxretries", 0); v > 0 {
 			m.maxRetries = v
 		}
+		if v := git.Int("lfs.transfer.maxretrydelay", -1); v > -1 {
+			m.maxRetryDelay = v
+		}
 		if v := git.Int("lfs.concurrenttransfers", 0); v > 0 {
 			m.concurrentTransfers = v
 		}
@@ -91,6 +101,9 @@ func NewManifest(f *fs.Filesystem, apiClient *lfsapi.Client, operation, remote s
 	if m.maxRetries < 1 {
 		m.maxRetries = defaultMaxRetries
 	}
+	if m.maxRetryDelay < 1 {
+		m.maxRetryDelay = defaultMaxRetryDelay
+	}
 
 	if m.concurrentTransfers < 1 {
 		m.concurrentTransfers = defaultConcurrentTransfers

tq/transfer_queue.go

@@ -17,37 +17,36 @@ import (
 
 const (
 	defaultBatchSize = 100
+	baseRetryDelayMs = 250
 )
 
 type retryCounter struct {
-	MaxRetries int `git:"lfs.transfer.maxretries"`
+	MaxRetries    int
+	MaxRetryDelay int
 
 	// cmu guards count
 	cmu sync.Mutex
 	// count maps OIDs to number of retry attempts
 	count map[string]int
 }
 
-// newRetryCounter instantiates a new *retryCounter. It parses the gitconfig
-// value: `lfs.transfer.maxretries`, and falls back to defaultMaxRetries if none
-// was provided.
-//
-// If it encountered an error in Unmarshaling the *config.Configuration, it will
-// be returned, otherwise nil.
+// newRetryCounter instantiates a new *retryCounter.
 func newRetryCounter() *retryCounter {
 	return &retryCounter{
-		MaxRetries: defaultMaxRetries,
-		count:      make(map[string]int),
+		MaxRetries:    defaultMaxRetries,
+		MaxRetryDelay: defaultMaxRetryDelay,
+		count:         make(map[string]int),
 	}
 }
 
-// Increment increments the number of retries for a given OID. It is safe to
-// call across multiple goroutines.
-func (r *retryCounter) Increment(oid string) {
+// Increment increments the number of retries for a given OID and returns the
+// new value. It is safe to call across multiple goroutines.
+func (r *retryCounter) Increment(oid string) int {
 	r.cmu.Lock()
 	defer r.cmu.Unlock()
 
 	r.count[oid]++
+	return r.count[oid]
 }
 
 // CountFor returns the current number of retries for a given OID. It is safe to
@@ -66,6 +65,22 @@ func (r *retryCounter) CanRetry(oid string) (int, bool) {
 	return count, count < r.MaxRetries
 }
 
+// ReadyTime returns the time from now when the current retry can occur or the
+// zero time if the retry can occur immediately.
+func (r *retryCounter) ReadyTime(oid string) time.Time {
+	count := r.CountFor(oid)
+	if count < 1 {
+		return time.Time{}
+	}
+
+	maxDelayMs := 1000 * uint64(r.MaxRetryDelay)
+	delay := uint64(baseRetryDelayMs) * (1 << uint(count-1))
+	if delay == 0 || delay > maxDelayMs {
+		delay = maxDelayMs
+	}
+	return time.Now().Add(time.Duration(delay) * time.Millisecond)
+}
+
 // batch implements the sort.Interface interface and enables sorting on a slice
 // of `*Transfer`s by object size.
 //
@@ -295,6 +310,7 @@ func NewTransferQueue(dir Direction, manifest *Manifest, remote string, options
 	}
 
 	q.rc.MaxRetries = q.manifest.maxRetries
+	q.rc.MaxRetryDelay = q.manifest.maxRetryDelay
 	q.client.MaxRetries = q.manifest.maxRetries
 
 	if q.batchSize <= 0 {
@@ -506,6 +522,24 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch batch) (batch, error)
 	next := q.makeBatch()
 	tracerx.Printf("tq: sending batch of size %d", len(batch))
 
+	enqueueRetry := func(t *objectTuple, err error, readyTime *time.Time) {
+		count := q.rc.Increment(t.Oid)
+
+		if readyTime == nil {
+			t.ReadyTime = q.rc.ReadyTime(t.Oid)
+		} else {
+			t.ReadyTime = *readyTime
+		}
+		delay := time.Until(t.ReadyTime).Seconds()
+
+		var errMsg string
+		if err != nil {
+			errMsg = fmt.Sprintf(": %s", err)
+		}
+		tracerx.Printf("tq: enqueue retry #%d after %.2fs for %q (size: %d)%s", count, delay, t.Oid, t.Size, errMsg)
+		next = append(next, t)
+	}
+
 	q.meter.Pause()
 	var bRes *BatchResponse
 	if q.manifest.standaloneTransferAgent != "" {
@@ -530,14 +564,10 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch batch) (batch, error)
 			// retried, they will be marked as failed.
 			for _, t := range batch {
 				if q.canRetryObject(t.Oid, err) {
-					q.rc.Increment(t.Oid)
-
-					next = append(next, t)
+					enqueueRetry(t, err, nil)
 				} else if readyTime, canRetry := q.canRetryObjectLater(t.Oid, err); canRetry {
-					tracerx.Printf("tq: retrying object %s after %s seconds.", t.Oid, time.Until(readyTime).Seconds())
 					err = nil
-					t.ReadyTime = readyTime
-					next = append(next, t)
+					enqueueRetry(t, err, &readyTime)
 				} else {
 					q.wait.Done()
 				}
@@ -599,13 +629,8 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch batch) (batch, error)
 			tr := newTransfer(o, objects.First().Name, objects.First().Path)
 
 			if a, err := tr.Rel(q.direction.String()); err != nil {
-				// XXX(taylor): duplication
 				if q.canRetryObject(tr.Oid, err) {
-					q.rc.Increment(tr.Oid)
-					count := q.rc.CountFor(tr.Oid)
-
-					tracerx.Printf("tq: enqueue retry #%d for %q (size: %d): %s", count, tr.Oid, tr.Size, err)
-					next = append(next, objects.First())
+					enqueueRetry(objects.First(), err, nil)
 				} else {
 					q.errorc <- errors.Errorf("[%v] %v", tr.Name, err)
 
@@ -624,12 +649,7 @@ func (q *TransferQueue) enqueueAndCollectRetriesFor(batch batch) (batch, error)
 
 	retries := q.addToAdapter(bRes.endpoint, toTransfer)
 	for t := range retries {
-		q.rc.Increment(t.Oid)
-		count := q.rc.CountFor(t.Oid)
-
-		tracerx.Printf("tq: enqueue retry #%d for %q (size: %d)", count, t.Oid, t.Size)
-
-		next = append(next, t)
+		enqueueRetry(t, nil, nil)
 	}
 
 	return next, nil

tq/transfer_queue_test.go

@@ -2,6 +2,7 @@ package tq
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -10,15 +11,27 @@ func TestManifestDefaultsToFixedRetries(t *testing.T) {
 	assert.Equal(t, 8, NewManifest(nil, nil, "", "").MaxRetries())
 }
 
+func TestManifestDefaultsToFixedRetryDelay(t *testing.T) {
+	assert.Equal(t, 10, NewManifest(nil, nil, "", "").MaxRetryDelay())
+}
+
 func TestRetryCounterDefaultsToFixedRetries(t *testing.T) {
 	rc := newRetryCounter()
 	assert.Equal(t, 8, rc.MaxRetries)
 }
 
+func TestRetryCounterDefaultsToFixedRetryDelay(t *testing.T) {
+	rc := newRetryCounter()
+	assert.Equal(t, 10, rc.MaxRetryDelay)
+}
+
 func TestRetryCounterIncrementsObjects(t *testing.T) {
 	rc := newRetryCounter()
-	rc.Increment("oid")
+	assert.Equal(t, 1, rc.Increment("oid"))
 	assert.Equal(t, 1, rc.CountFor("oid"))
+
+	assert.Equal(t, 2, rc.Increment("oid"))
+	assert.Equal(t, 2, rc.CountFor("oid"))
 }
 
 func TestRetryCounterCanNotRetryAfterExceedingRetryCount(t *testing.T) {
@@ -31,6 +44,36 @@ func TestRetryCounterCanNotRetryAfterExceedingRetryCount(t *testing.T) {
 	assert.False(t, canRetry)
 }
 
+func TestRetryCounterDoesNotDelayFirstAttempt(t *testing.T) {
+	rc := newRetryCounter()
+	assert.Equal(t, time.Time{}, rc.ReadyTime("oid"))
+}
+
+func TestRetryCounterDelaysExponentially(t *testing.T) {
+	rc := newRetryCounter()
+	start := time.Now()
+
+	rc.Increment("oid")
+	ready1 := rc.ReadyTime("oid")
+	assert.GreaterOrEqual(t, int64(ready1.Sub(start)/time.Millisecond), int64(baseRetryDelayMs))
+
+	rc.Increment("oid")
+	ready2 := rc.ReadyTime("oid")
+	assert.GreaterOrEqual(t, int64(ready2.Sub(start)/time.Millisecond), int64(2*baseRetryDelayMs))
+}
+
+func TestRetryCounterLimitsDelay(t *testing.T) {
+	rc := newRetryCounter()
+	rc.MaxRetryDelay = 1
+
+	for i := 0; i < 4; i++ {
+		rc.Increment("oid")
+	}
+
+	rt := rc.ReadyTime("oid")
+	assert.WithinDuration(t, time.Now(), rt, 1*time.Second)
+}
+
 func TestBatchSizeReturnsBatchSize(t *testing.T) {
 	q := NewTransferQueue(
 		Upload, NewManifest(nil, nil, "", ""), "origin", WithBatchSize(3))