feat(dlq): retry failed tasks from the dead-letter queue

core/cmd/admin/queue_backup.go

@@ -101,10 +101,50 @@ var listFailedBackupCmd = &cobra.Command{
 	},
 }
 
+//nolint:gochecknoglobals
+var retryBackupCmd = &cobra.Command{
+	Use:   "retry [cluster-name]",
+	Short: "Retry failed backup tasks in the queue",
+	Long: "Retry failed backup tasks in the queue.\n\n" +
+		"With no arguments, all failed backup tasks are retried. If a cluster name " +
+		"is given, all failed backup tasks for that cluster are retried.",
+	Args: cobra.MaximumNArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		socketPath, err := cmd.Flags().GetString("socket-path")
+		if err != nil {
+			return fmt.Errorf("while getting the socketPath flag: %w", err)
+		}
+
+		conn, err := connectToAdminServer(socketPath)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			_ = conn.Close()
+		}()
+
+		var request klioGRPC.QueueRetryBackupsRequest
+		if len(args) > 0 {
+			clusterName := args[0]
+			request.ClusterName = &clusterName
+		}
+
+		adminClient := klioGRPC.NewAdminClient(conn)
+		_, err = adminClient.QueueRetryBackups(cmd.Context(), &request)
+		if err != nil {
+			return fmt.Errorf("while calling queue retry backups entrypoint: %w", err)
+		}
+
+		return nil
+	},
+}
+
 //nolint:gochecknoinits
 func init() {
 	queueCmd.AddCommand(queueBackupCmd)
-	queueBackupCmd.AddCommand(listFailedBackupCmd)
 
+	queueBackupCmd.AddCommand(listFailedBackupCmd)
 	listFailedBackupCmd.Flags().String("cluster-name", "", "Cluster name to filter failed backup tasks (optional)")
+
+	queueBackupCmd.AddCommand(retryBackupCmd)
 }

core/cmd/admin/queue_wal.go

@@ -22,7 +22,6 @@ package admin
 import (
 	"fmt"
 	"os"
-	"strconv"
 	"time"
 
 	"github.com/spf13/cobra"
@@ -89,15 +88,14 @@ var listFailedWALCmd = &cobra.Command{
 			rows := make([][]string, 0, len(response.GetWals()))
 			for _, wal := range response.GetWals() {
 				rows = append(rows, []string{
-					strconv.FormatUint(wal.GetSequence(), 10),
 					wal.GetClusterName(),
 					wal.GetWalName(),
 					wal.GetLastAttemptTime().AsTime().Format(time.RFC3339),
 				})
 			}
 			if err := writeTable(
 				os.Stdout,
-				[]string{"SEQUENCE", "CLUSTER", "WAL NAME", "LAST ATTEMPT"},
+				[]string{"CLUSTER", "WAL NAME", "LAST ATTEMPT"},
 				rows,
 			); err != nil {
 				return fmt.Errorf("while writing table output: %w", err)
@@ -108,10 +106,54 @@ var listFailedWALCmd = &cobra.Command{
 	},
 }
 
+//nolint:gochecknoglobals
+var retryWALCmd = &cobra.Command{
+	Use:   "retry [cluster-name] [WAL1 WAL2 ...]",
+	Short: "Retry failed WAL tasks in the queue",
+	Long: "Retry failed WAL tasks in the queue.\n\n" +
+		"With no arguments, all failed WAL tasks are retried. If a cluster name " +
+		"is given, all failed WAL tasks for that cluster are retried. If WAL " +
+		"files are also given, only those are retried.",
+	Args: cobra.ArbitraryArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		socketPath, err := cmd.Flags().GetString("socket-path")
+		if err != nil {
+			return fmt.Errorf("while getting the socketPath flag: %w", err)
+		}
+
+		conn, err := connectToAdminServer(socketPath)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			_ = conn.Close()
+		}()
+
+		var request klioGRPC.QueueRetryWALsRequest
+		if len(args) > 0 {
+			clusterName := args[0]
+			request.ClusterName = &clusterName
+		}
+		if len(args) > 1 {
+			request.WalNames = args[1:]
+		}
+
+		adminClient := klioGRPC.NewAdminClient(conn)
+		_, err = adminClient.QueueRetryWALs(cmd.Context(), &request)
+		if err != nil {
+			return fmt.Errorf("while calling queue retry wals entrypoint: %w", err)
+		}
+
+		return nil
+	},
+}
+
 //nolint:gochecknoinits
 func init() {
 	queueCmd.AddCommand(queueWALCmd)
-	queueWALCmd.AddCommand(listFailedWALCmd)
 
+	queueWALCmd.AddCommand(listFailedWALCmd)
 	listFailedWALCmd.Flags().String("cluster-name", "", "Cluster name to filter failed WAL tasks (optional)")
+
+	queueWALCmd.AddCommand(retryWALCmd)
 }