diff --git a/database/etcddb/batch.go b/database/etcddb/batch.go index 24b83de9..3b7d1658 100644 --- a/database/etcddb/batch.go +++ b/database/etcddb/batch.go @@ -1,8 +1,16 @@ package etcddb import ( + "context" + "fmt" + "math" + "time" + "github.com/aptly-dev/aptly/database" + "github.com/rs/zerolog/log" clientv3 "go.etcd.io/etcd/client/v3" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type EtcDBatch struct { @@ -30,21 +38,112 @@ func (b *EtcDBatch) Write() (err error) { batchSize := 128 for i := 0; i < len(b.ops); i += batchSize { - txn := kv.Txn(Ctx) end := i + batchSize if end > len(b.ops) { end = len(b.ops) } batch := b.ops[i:end] - txn.Then(batch...) - _, err = txn.Commit() - if err != nil { - panic(err) + + // Retry logic with exponential backoff + var lastErr error + for retry := 0; retry <= DefaultWriteRetries; retry++ { + ctx, cancel := context.WithTimeout(context.Background(), DefaultTimeout) + txn := kv.Txn(ctx) + txn.Then(batch...) + _, err = txn.Commit() + cancel() + + if err == nil { + // Success, move to next batch + break + } + + lastErr = err + + // Check if error is retryable + if !isRetryableError(err) { + log.Error().Err(err).Int("batch_start", i).Int("batch_end", end).Msg("etcd: non-retryable error during batch write") + return fmt.Errorf("etcd batch write failed: %w", err) + } + + if retry < DefaultWriteRetries { + // Calculate exponential backoff + backoff := time.Duration(math.Pow(2, float64(retry))) * 100 * time.Millisecond + if backoff > 5*time.Second { + backoff = 5 * time.Second + } + + log.Warn().Err(err). + Int("retry", retry+1). + Int("max_retries", DefaultWriteRetries). + Dur("backoff", backoff). + Int("batch_start", i). + Int("batch_end", end). + Msg("etcd: batch write failed, retrying") + + time.Sleep(backoff) + } + } + + // All retries exhausted + if lastErr != nil { + log.Error().Err(lastErr). + Int("batch_start", i). + Int("batch_end", end). + Int("retries", DefaultWriteRetries). + Msg("etcd: batch write failed after all retries") + return fmt.Errorf("etcd batch write failed after %d retries: %w", DefaultWriteRetries, lastErr) } } - return + return nil +} + +// isRetryableError checks if an error is retryable +func isRetryableError(err error) bool { + if err == nil { + return false + } + + // Check for gRPC status errors + if s, ok := status.FromError(err); ok { + switch s.Code() { + case codes.Unavailable, codes.DeadlineExceeded, codes.ResourceExhausted, codes.Aborted: + return true + } + } + + // Check for context errors + if err == context.DeadlineExceeded || err == context.Canceled { + return true + } + + // Check for timeout errors in error message + if errStr := err.Error(); errStr != "" { + if contains(errStr, "timeout") || contains(errStr, "timed out") || + contains(errStr, "unavailable") || contains(errStr, "connection refused") { + return true + } + } + + return false +} + +// contains is a simple string contains helper +func contains(s, substr string) bool { + return len(substr) > 0 && len(s) >= len(substr) && + (s == substr || s[:len(substr)] == substr || s[len(s)-len(substr):] == substr || + len(s) > len(substr) && findSubstring(s, substr)) +} + +func findSubstring(s, substr string) bool { + for i := 0; i <= len(s)-len(substr); i++ { + if s[i:i+len(substr)] == substr { + return true + } + } + return false } // batch should implement database.Batch