Retry SyncExec (#496)

* Retry SyncExec

* Update to use httputil backoff

* Remove attempts

* Redundant params

Author: hweawer

build-index/tagstore/store.go

@@ -59,6 +59,10 @@ type tagStore struct {
 	fs               FileStore
 	backends         *backend.Manager
 	writeBackManager persistedretry.Manager
+
+	// writeBackStrategy determines how tags are written to backend storage.
+	// Set at initialization based on WriteThrough config.
+	writeBackStrategy func(task persistedretry.Task) error
 }
 
 // New creates a new Store.
@@ -73,42 +77,33 @@ func New(
 		"module": "tagstore",
 	})
 
-	return &tagStore{
+	s := &tagStore{
 		config:           config,
 		fs:               fs,
 		backends:         backends,
 		writeBackManager: writeBackManager,
 	}
+
+	// Set write-back strategy based on configuration
+	if config.WriteThrough {
+		s.writeBackStrategy = s.writeThroughStrategy
+	} else {
+		s.writeBackStrategy = s.asyncWriteBackStrategy
+	}
+
+	return s
 }
 
 func (s *tagStore) Put(tag string, d core.Digest, writeBackDelay time.Duration) error {
-	log.With("tag", tag, "digest", d.String(), "writeback_delay", writeBackDelay).Debug("Storing tag to disk")
-
 	if err := s.writeTagToDisk(tag, d); err != nil {
-		log.With("tag", tag, "digest", d.String(), "error", err).Error("Failed to write tag to disk")
 		return fmt.Errorf("write tag to disk: %s", err)
 	}
 	if _, err := s.fs.SetCacheFileMetadata(tag, metadata.NewPersist(true)); err != nil {
-		log.With("tag", tag, "digest", d.String(), "error", err).Error("Failed to set persist metadata")
 		return fmt.Errorf("set persist metadata: %s", err)
 	}
 
 	task := writeback.NewTask(tag, tag, writeBackDelay)
-	if s.config.WriteThrough {
-		log.With("tag", tag, "digest", d.String()).Debug("Using write-through mode for tag")
-		if err := s.writeBackManager.SyncExec(task); err != nil {
-			log.With("tag", tag, "digest", d.String(), "error", err).Error("Failed to sync write-back tag to backend")
-			return fmt.Errorf("sync exec write-back task: %s", err)
-		}
-		log.With("tag", tag, "digest", d.String()).Info("Tag written to backend synchronously")
-	} else {
-		log.With("tag", tag, "digest", d.String(), "writeback_delay", writeBackDelay).Debug("Adding async write-back task for tag")
-		if err := s.writeBackManager.Add(task); err != nil {
-			log.With("tag", tag, "digest", d.String(), "error", err).Error("Failed to add write-back task")
-			return fmt.Errorf("add write-back task: %s", err)
-		}
-	}
-	return nil
+	return s.writeBackStrategy(task)
 }
 
 func (s *tagStore) Get(tag string) (d core.Digest, err error) {
@@ -125,6 +120,22 @@ func (s *tagStore) Get(tag string) (d core.Digest, err error) {
 	return d, err
 }
 
+// writeThroughStrategy writes tags synchronously to backend storage.
+func (s *tagStore) writeThroughStrategy(task persistedretry.Task) error {
+	if err := s.writeBackManager.SyncExec(task); err != nil {
+		return fmt.Errorf("sync exec write-back task: %s", err)
+	}
+	return nil
+}
+
+// asyncWriteBackStrategy queues tags for asynchronous write-back to backend storage.
+func (s *tagStore) asyncWriteBackStrategy(task persistedretry.Task) error {
+	if err := s.writeBackManager.Add(task); err != nil {
+		return fmt.Errorf("add write-back task: %s", err)
+	}
+	return nil
+}
+
 func (s *tagStore) writeTagToDisk(tag string, d core.Digest) error {
 	buf := bytes.NewBufferString(d.String())
 	if err := s.fs.CreateCacheFile(tag, buf); err != nil && !os.IsExist(err) {

lib/persistedretry/config.go

@@ -13,7 +13,11 @@
 // limitations under the License.
 package persistedretry
 
-import "time"
+import (
+	"time"
+
+	"github.com/uber/kraken/utils/httputil"
+)
 
 // Config defines Manager configuration.
 type Config struct {
@@ -32,6 +36,9 @@ type Config struct {
 	// Interval at which retries should be polled from storage.
 	PollRetriesInterval time.Duration `yaml:"poll_retries_interval"`
 
+	// Exponential backoff configuration for synchronous task execution.
+	SyncRetryBackoff httputil.ExponentialBackOffConfig `yaml:"sync_retry_backoff"`
+
 	// Flags that zero-value channel sizes should not have defaults applied.
 	Testing bool
 }
@@ -52,6 +59,17 @@ func (c Config) applyDefaults() Config {
 	if c.RetryInterval == 0 {
 		c.RetryInterval = 30 * time.Second
 	}
+	if !c.SyncRetryBackoff.Enabled {
+		// Apply defaults for sync retry backoff to maintain backward compatibility
+		c.SyncRetryBackoff = httputil.ExponentialBackOffConfig{
+			Enabled:             true,
+			InitialInterval:     500 * time.Millisecond,
+			RandomizationFactor: 0.05,
+			Multiplier:          2,
+			MaxInterval:         30 * time.Second,
+			MaxRetries:          2, // 2 retries = 3 total attempts (1 initial + 2 retries)
+		}
+	}
 	if !c.Testing {
 		if c.IncomingBuffer == 0 {
 			c.IncomingBuffer = 1000

lib/persistedretry/manager.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cenkalti/backoff"
 	"github.com/uber-go/tally"
 	"go.uber.org/atomic"
 
@@ -54,8 +55,8 @@ type manager struct {
 
 // NewManager creates a new Manager.
 func NewManager(
-	config Config, stats tally.Scope, store Store, executor Executor) (Manager, error) {
-
+	config Config, stats tally.Scope, store Store, executor Executor,
+) (Manager, error) {
 	stats = stats.Tagged(map[string]string{
 		"module":   "persistedretry",
 		"executor": executor.Name(),
@@ -143,10 +144,20 @@ func (m *manager) Add(t Task) error {
 	return nil
 }
 
-// SyncExec executes the task synchronously.
-// Tasks will NOT be added to the retry queue if fail.
+// SyncExec executes the task synchronously with retry logic.
+// Tasks will NOT be added to the retry queue if fail, but will be retried
+// in-place according to the configured SyncRetryBackoff.
 func (m *manager) SyncExec(t Task) error {
-	return m.executor.Exec(t)
+	bo := m.config.SyncRetryBackoff.Build()
+
+	operation := func() error {
+		return m.executor.Exec(t)
+	}
+
+	if err := backoff.Retry(operation, bo); err != nil {
+		return fmt.Errorf("sync task failed: %w", err)
+	}
+	return nil
 }
 
 // Close waits for all workers to exit current task.