Merge pull request #7707 from ellemouton/wtclientFixes

watchtower: miscellaneous fixes

watchtower/blob/type.go

@@ -67,6 +67,20 @@ const (
 	TypeRewardCommit = Type(FlagCommitOutputs | FlagReward)
 )
 
+// Identifier returns a unique, stable string identifier for the blob Type.
+func (t Type) Identifier() (string, error) {
+	switch t {
+	case TypeAltruistCommit:
+		return "legacy", nil
+	case TypeAltruistAnchorCommit:
+		return "anchor", nil
+	case TypeRewardCommit:
+		return "reward", nil
+	default:
+		return "", fmt.Errorf("unknown blob type: %v", t)
+	}
+}
+
 // Has returns true if the Type has the passed flag enabled.
 func (t Type) Has(flag Flag) bool {
 	return Flag(t)&flag == flag

watchtower/wtclient/client.go

@@ -350,10 +350,12 @@ func New(config *Config) (*TowerClient, error) {
 		cfg.WriteTimeout = DefaultWriteTimeout
 	}
 
-	prefix := "(legacy)"
-	if cfg.Policy.IsAnchorChannel() {
-		prefix = "(anchor)"
+	identifier, err := cfg.Policy.BlobType.Identifier()
+	if err != nil {
+		return nil, err
 	}
+	prefix := fmt.Sprintf("(%s)", identifier)
+
 	plog := build.NewPrefixLog(prefix, log)
 
 	// Load the sweep pkscripts that have been generated for all previously
@@ -363,10 +365,7 @@ func New(config *Config) (*TowerClient, error) {
 		return nil, err
 	}
 
-	var (
-		policy  = cfg.Policy.BlobType.String()
-		queueDB = cfg.DB.GetDBQueue([]byte(policy))
-	)
+	queueDB := cfg.DB.GetDBQueue([]byte(identifier))
 	queue, err := NewDiskOverflowQueue[*wtdb.BackupID](
 		queueDB, cfg.MaxTasksInMemQueue, plog,
 	)
@@ -678,7 +677,11 @@ func (c *TowerClient) Start() error {
 
 		// Start the task pipeline to which new backup tasks will be
 		// submitted from active links.
-		c.pipeline.Start()
+		err = c.pipeline.Start()
+		if err != nil {
+			returnErr = err
+			return
+		}
 
 		c.wg.Add(1)
 		go c.backupDispatcher()
@@ -727,7 +730,10 @@ func (c *TowerClient) Stop() error {
 
 		// 4. Since all valid tasks have been assigned to session
 		// queues, we no longer need to negotiate sessions.
-		c.negotiator.Stop()
+		err = c.negotiator.Stop()
+		if err != nil {
+			returnErr = err
+		}
 
 		c.log.Debugf("Waiting for active session queues to finish "+
 			"draining, stats: %s", c.stats)