gazctl shards prune: retry transient fragment removal errors

cmd/gazctl/gazctlcmd/shards_prune.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 	"go.gazette.dev/core/broker/client"
@@ -157,7 +158,7 @@ func (cmd *cmdShardsPrune) Execute([]string) error {
 			group.Go(func() error {
 				var err error
 				if !cmd.DryRun {
-					err = fragment.Remove(ctx, spec)
+					err = removeFragment(ctx, spec)
 				}
 
 				mu.Lock()
@@ -206,6 +207,33 @@ func (cmd *cmdShardsPrune) Execute([]string) error {
 	return nil
 }
 
+const removeAttempts = 3
+
+var removeRetryInterval = time.Second
+
+// removeFragment deletes `spec` from its store, retrying errors which are
+// commonly transient: 5xx responses, rate limits, and dropped connections.
+func removeFragment(ctx context.Context, spec pb.Fragment) error {
+	for attempt := 0; ; attempt++ {
+		var err = fragment.Remove(ctx, spec)
+
+		if err == nil {
+			return nil
+		} else if fragment.IsAuthError(spec, err) || ctx.Err() != nil {
+			return err
+		} else if attempt+1 == removeAttempts {
+			return err
+		}
+
+		log.WithFields(log.Fields{
+			"fragment": spec,
+			"error":    err,
+		}).Warn("failed to remove fragment (will retry)")
+
+		time.Sleep(removeRetryInterval)
+	}
+}
+
 // checkRecoveryLogStoresHealth checks if all fragment stores for the given recovery log journal are healthy.
 // Returns true if all stores are healthy, false otherwise.
 func checkRecoveryLogStoresHealth(ctx context.Context, jc pb.JournalClient, recoveryLog pb.Journal) bool {

cmd/gazctl/gazctlcmd/shards_prune_test.go

@@ -1,11 +1,16 @@
 package gazctlcmd
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
+	"net/url"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	pb "go.gazette.dev/core/broker/protocol"
+	"go.gazette.dev/core/broker/stores"
 	pc "go.gazette.dev/core/consumer/protocol"
 	"go.gazette.dev/core/consumer/recoverylog"
 )
@@ -470,3 +475,66 @@ func TestSegmentFoldingWithManyLogs(t *testing.T) {
 		},
 	}, m)
 }
+
+func TestRemoveFragment(t *testing.T) {
+	defer func(i time.Duration) { removeRetryInterval = i }(removeRetryInterval)
+	removeRetryInterval = 0
+
+	var authError = errors.New("access denied")
+	var transientError = errors.New("connection reset by peer")
+	var attempts = make(map[string]int)
+
+	// A store whose deletes fail with `err` until `failures` are exhausted.
+	var provider = func(err error, failures int) stores.Constructor {
+		return func(u *url.URL) (stores.Store, error) {
+			return &stores.CallbackStore{
+				Fallback: stores.NewMemoryStore(u),
+				RemoveFunc: func(fallback stores.Store, ctx context.Context, path string) error {
+					if attempts[u.Scheme]++; attempts[u.Scheme] <= failures {
+						return err
+					}
+					return fallback.Remove(ctx, path)
+				},
+				IsAuthErrorFunc: func(_ stores.Store, e error) bool {
+					return e == authError
+				},
+			}, nil
+		}
+	}
+
+	stores.RegisterProviders(map[string]stores.Constructor{
+		"s3":   provider(transientError, 1),  // A transient failure, then success.
+		"gs":   provider(transientError, 99), // Always fails.
+		"file": provider(authError, 99),      // Denies deletes.
+	})
+
+	var frag = pb.Fragment{
+		Journal:          "a/log",
+		Begin:            0,
+		End:              100,
+		CompressionCodec: pb.CompressionCodec_NONE,
+	}
+
+	// A transient error is retried until the delete succeeds.
+	frag.BackingStore = "s3://bucket/"
+	require.NoError(t, removeFragment(context.Background(), frag))
+	require.Equal(t, 2, attempts["s3"])
+
+	// Retries are bounded, and the final error is returned.
+	frag.BackingStore = "gs://bucket/"
+	require.Equal(t, transientError, removeFragment(context.Background(), frag))
+	require.Equal(t, removeAttempts, attempts["gs"])
+
+	// An authorization error is returned on its first occurrence.
+	frag.BackingStore = "file:///root/"
+	require.Equal(t, authError, removeFragment(context.Background(), frag))
+	require.Equal(t, 1, attempts["file"])
+
+	// A cancelled prune stops retrying.
+	var ctx, cancel = context.WithCancel(context.Background())
+	cancel()
+
+	attempts["gs"], frag.BackingStore = 0, "gs://bucket/"
+	require.Equal(t, transientError, removeFragment(ctx, frag))
+	require.Equal(t, 1, attempts["gs"])
+}