diff --git a/cmd/gazctl/gazctlcmd/shards_prune.go b/cmd/gazctl/gazctlcmd/shards_prune.go index afb4ab2a..66b49d53 100644 --- a/cmd/gazctl/gazctlcmd/shards_prune.go +++ b/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 { diff --git a/cmd/gazctl/gazctlcmd/shards_prune_test.go b/cmd/gazctl/gazctlcmd/shards_prune_test.go index 3b7f1ced..7ae004d1 100644 --- a/cmd/gazctl/gazctlcmd/shards_prune_test.go +++ b/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"]) +}