Sweep fallback across primary and fallback arkd-wallets

README.md

@@ -189,7 +189,7 @@ export ARKD_WALLET_ADDR=localhost:6060
 export ARKD_WALLET_FALLBACK_ADDRS=localhost:6061,localhost:6062
 ```
 
-Every wallet, primary and fallback, must be initialized and unlocked out of band (see [Setup arkd](#setup-arkd)) and must be on the same network as the primary; `arkd` validates this at startup and refuses to start otherwise. The primary wallet remains the sole source of the forfeit address, connector address, scanning and signing. The additional wallets are used only as sweep fallbacks; that wiring lands in a later change.
+Every wallet, primary and fallback, must be initialized and unlocked out of band (see [Setup arkd](#setup-arkd)) and must be on the same network as the primary; `arkd` validates this at startup and refuses to start otherwise. The primary wallet remains the sole source of the forfeit address, connector address, scanning and signing. The additional wallets are used only as sweep fallbacks: when a batch cannot be signed by the primary wallet (for example a batch created before this wallet became the primary), `arkd` tries each fallback wallet in turn until one can sign it. The swept funds always go to the primary wallet's address, whichever wallet signs.
 
 ### Connect to signer
 

docker-compose.regtest.yml

@@ -61,9 +61,9 @@ services:
       - ARKD_WALLET_DEPRECATED_SIGNER_KEYS=${ARKD_WALLET_DEPRECATED_SIGNER_KEYS:-}
     volumes:
       - arkd-wallet-volume:/app/data
-  # A second arkd-wallet acting as an additional LP wallet, registered with arkd
-  # via ARKD_WALLET_FALLBACK_ADDRS. arkd currently only dials it and verifies it
-  # is initialized and unlocked at startup; it is not yet used to sign sweeps.
+  # A second arkd-wallet acting as an additional LP wallet, wired into arkd as a
+  # sweep fallback via ARKD_WALLET_FALLBACK_ADDRS: sweep signing is attempted with
+  # the primary first, then each fallback.
   arkd-wallet-2:
     restart: unless-stopped
     build:

internal/config/config.go

@@ -680,6 +680,17 @@ func (c *Config) FallbackWallets() []FallbackWallet {
 	return c.walletFallbacks
 }
 
+// fallbackWalletServices returns just the dialed fallback wallet clients, for
+// consumers (the app and admin services) that sign with them but don't need the
+// dial address.
+func (c *Config) fallbackWalletServices() []ports.WalletService {
+	svcs := make([]ports.WalletService, 0, len(c.walletFallbacks))
+	for _, fb := range c.walletFallbacks {
+		svcs = append(svcs, fb.Service)
+	}
+	return svcs
+}
+
 func (c *Config) UnlockerService() ports.Unlocker {
 	return c.unlocker
 }
@@ -842,13 +853,9 @@ func (c *Config) walletService() error {
 	return nil
 }
 
-// dialFallbackWallets dials the configured fallback arkd-wallets and validates
-// that each one is reachable and on the same network as the primary. Fallback
-// wallets belong to additional liquidity providers and are intended as sweep
-// fallbacks, though arkd does not yet use them to sign sweeps; for now they are
-// only dialed and readiness-checked. The primary remains the sole source of the
-// forfeit pubkey, addresses and signing. Any failure is fatal so a misconfigured
-// wallet is surfaced at startup.
+// dialFallbackWallets dials the configured fallback arkd-wallets, used as sweep
+// signers, validating each is reachable and on the primary's network. Any
+// failure is fatal so misconfiguration surfaces at startup.
 func (c *Config) dialFallbackWallets() ([]FallbackWallet, error) {
 	fallbacks := make([]FallbackWallet, 0, len(c.WalletFallbackAddrs))
 	seen := make(map[string]struct{}, len(c.WalletFallbackAddrs))
@@ -1008,7 +1015,7 @@ func (c *Config) appService() error {
 	}
 
 	svc, err := application.NewService(
-		c.wallet, c.signer, c.repo, c.txBuilder, c.scanner,
+		c.wallet, c.fallbackWalletServices(), c.signer, c.repo, c.txBuilder, c.scanner,
 		c.scheduler, c.liveStore, roundReportSvc, c.alerts, c.fee,
 	)
 	if err != nil {
@@ -1026,7 +1033,7 @@ func (c *Config) adminService() error {
 	}
 
 	c.adminSvc = application.NewAdminService(
-		c.wallet, c.repo, c.txBuilder, c.liveStore, unit, c.fee,
+		c.wallet, c.fallbackWalletServices(), c.repo, c.txBuilder, c.liveStore, unit, c.fee,
 	)
 	return nil
 }

internal/core/application/admin.go

@@ -59,7 +59,10 @@ type AdminService interface {
 }
 
 type adminService struct {
-	walletSvc       ports.WalletService
+	walletSvc ports.WalletService
+	// walletFallbacks are additional arkd-wallets whose batches admin.Sweep may also
+	// sign; signing is attempted with the primary wallet first, then each fallback.
+	walletFallbacks []ports.WalletService
 	repoManager     ports.RepoManager
 	txBuilder       ports.TxBuilder
 	sweeperTimeUnit ports.TimeUnit
@@ -72,11 +75,13 @@ type adminService struct {
 }
 
 func NewAdminService(
-	walletSvc ports.WalletService, repoManager ports.RepoManager, txBuilder ports.TxBuilder,
+	walletSvc ports.WalletService, walletFallbacks []ports.WalletService,
+	repoManager ports.RepoManager, txBuilder ports.TxBuilder,
 	liveStoreSvc ports.LiveStore, timeUnit ports.TimeUnit, feeManager ports.FeeManager,
 ) AdminService {
 	return &adminService{
 		walletSvc:       walletSvc,
+		walletFallbacks: walletFallbacks,
 		repoManager:     repoManager,
 		txBuilder:       txBuilder,
 		sweeperTimeUnit: timeUnit,
@@ -85,6 +90,12 @@ func NewAdminService(
 	}
 }
 
+// signingWallets returns the wallets to try when signing a sweep, in order: the
+// primary wallet first, then any configured fallbacks.
+func (a *adminService) signingWallets() []ports.WalletService {
+	return append([]ports.WalletService{a.walletSvc}, a.walletFallbacks...)
+}
+
 func (a *adminService) Wallet() ports.WalletService {
 	return a.walletSvc
 }
@@ -583,7 +594,7 @@ func (a *adminService) Sweep(
 		)
 	}
 
-	txid, txhex, err = a.txBuilder.BuildSweepTx(inputs)
+	txid, txhex, err = buildAndSignSweepTx(a.txBuilder, a.signingWallets(), inputs)
 	if err != nil {
 		return
 	}

internal/core/application/admin_test.go

@@ -56,7 +56,7 @@ func TestAdminService_Settings(t *testing.T) {
 		if seed != nil {
 			require.NoError(t, repo.settingsRepo.Upsert(ctx, *seed, nil))
 		}
-		return application.NewAdminService(nil, repo, nil, nil, ports.UnixTime, nil)
+		return application.NewAdminService(nil, nil, repo, nil, nil, ports.UnixTime, nil)
 	}
 
 	t.Run("settings", func(t *testing.T) {
@@ -246,7 +246,7 @@ func TestAdminService_SettingsSerialization(t *testing.T) {
 	seed := validSettings()
 	probe := &serializeProbeRepo{settings: &seed, delay: 10 * time.Millisecond}
 	svc := application.NewAdminService(
-		nil, &mockRepoManager{settingsRepo: probe}, nil, nil, ports.UnixTime, nil,
+		nil, nil, &mockRepoManager{settingsRepo: probe}, nil, nil, ports.UnixTime, nil,
 	)
 
 	const workers = 8

internal/core/application/service.go

@@ -69,6 +69,7 @@ type service struct {
 
 func NewService(
 	wallet ports.WalletService,
+	walletFallbacks []ports.WalletService,
 	signer ports.SignerService,
 	repoManager ports.RepoManager,
 	builder ports.TxBuilder,
@@ -142,13 +143,19 @@ func NewService(
 	ctx, cancel := context.WithCancel(ctx)
 
 	svc := &service{
-		wallet:                   wallet,
-		signer:                   signer,
-		repoManager:              repoManager,
-		builder:                  builder,
-		cache:                    cache,
-		scanner:                  scanner,
-		sweeper:                  newSweeper(wallet, repoManager, builder, scheduler),
+		wallet:      wallet,
+		signer:      signer,
+		repoManager: repoManager,
+		builder:     builder,
+		cache:       cache,
+		scanner:     scanner,
+		sweeper: newSweeper(
+			wallet,
+			walletFallbacks,
+			repoManager,
+			builder,
+			scheduler,
+		),
 		operatorPrvkey:           operatorSigningKey,
 		operatorPubkey:           operatorSigningKey.PubKey(),
 		forfeitsBoardingSigsChan: make(chan struct{}, 1),

internal/core/application/sweeper.go

@@ -31,10 +31,13 @@ type sweeperTask struct {
 // it is responsible for sweeping batch outputs that reached the expiration date.
 // it also handles delaying the sweep events in case some parts of the tree are broadcasted
 type sweeper struct {
-	wallet      ports.WalletService
-	repoManager ports.RepoManager
-	builder     ports.TxBuilder
-	scheduler   ports.SchedulerService
+	wallet ports.WalletService
+	// walletFallbacks are additional arkd-wallets whose batches this arkd may also
+	// sweep; signing is attempted with the primary wallet first, then each fallback.
+	walletFallbacks []ports.WalletService
+	repoManager     ports.RepoManager
+	builder         ports.TxBuilder
+	scheduler       ports.SchedulerService
 
 	// cache of scheduled tasks, avoid scheduling the same sweep event multiple times
 	locker *sync.Mutex
@@ -44,14 +47,64 @@ type sweeper struct {
 }
 
 func newSweeper(
-	wallet ports.WalletService, repoManager ports.RepoManager, builder ports.TxBuilder,
+	wallet ports.WalletService, walletFallbacks []ports.WalletService,
+	repoManager ports.RepoManager, builder ports.TxBuilder,
 	scheduler ports.SchedulerService,
 ) *sweeper {
 	return &sweeper{
-		wallet, repoManager, builder, scheduler, &sync.Mutex{}, make(map[string]struct{}), nil,
+		wallet:          wallet,
+		walletFallbacks: walletFallbacks,
+		repoManager:     repoManager,
+		builder:         builder,
+		scheduler:       scheduler,
+		locker:          &sync.Mutex{},
+		scheduledTasks:  make(map[string]struct{}),
 	}
 }
 
+// signingWallets returns the wallets to try when signing a sweep, in order: the
+// primary wallet first, then any configured fallbacks.
+func (s *sweeper) signingWallets() []ports.WalletService {
+	return append([]ports.WalletService{s.wallet}, s.walletFallbacks...)
+}
+
+// buildAndSignSweepTx builds the sweep transaction once (its destination and fees
+// come from the primary wallet) and then attempts to sign it with each wallet in
+// order, returning (txid, signedTx, nil) as soon as one succeeds. This lets a
+// single arkd sweep batches signed by any of its primary/fallback arkd-wallets.
+// Broadcasting is left to the caller.
+//
+// On signing failure it still returns the txid (with a non-nil error and an empty
+// signed tx) so a reconcile-only caller can use the txid without rebuilding the
+// tx. Callers that broadcast must treat a non-nil error as fatal. A build failure
+// returns an empty txid.
+func buildAndSignSweepTx(
+	builder ports.TxBuilder, wallets []ports.WalletService, inputs []ports.TxInput,
+) (string, string, error) {
+	txid, unsignedTx, err := builder.BuildSweepTx(inputs)
+	if err != nil {
+		return "", "", err
+	}
+
+	if len(wallets) == 0 {
+		return txid, "", fmt.Errorf("no signing wallets configured for sweep tx %s", txid)
+	}
+
+	signErrs := make([]error, 0, len(wallets))
+	for i, wallet := range wallets {
+		signed, signErr := builder.SignSweepTx(wallet, unsignedTx)
+		if signErr == nil {
+			return txid, signed, nil
+		}
+		// name the failing wallet so a multi-wallet operator can tell which rejected it
+		signErrs = append(signErrs, fmt.Errorf("wallet[%d]: %w", i, signErr))
+	}
+
+	return txid, "", fmt.Errorf(
+		"no wallet could sign sweep tx %s: %w", txid, errors.Join(signErrs...),
+	)
+}
+
 func (s *sweeper) start(ctx context.Context) error {
 	s.scheduledTasks = make(map[string]struct{})
 	s.scheduler.Start()
@@ -627,7 +680,9 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string)
 			)
 
 			// build the sweep transaction with all the expired non-swept batch outputs
-			sweepTxId, sweepTx, err = s.builder.BuildSweepTx(unspentOutputsToSweep)
+			sweepTxId, sweepTx, err = buildAndSignSweepTx(
+				s.builder, s.signingWallets(), unspentOutputsToSweep,
+			)
 			if err != nil {
 				return err
 			}
@@ -661,11 +716,26 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string)
 			}
 			log.Debugf("sweeper: batch %s swept by: %s", commitmentTxid, txid)
 		} else {
-			// if all outputs are spent, it means we missed to mark the batch as swept,
-			// build a sweep transaction without broadcasting it. we'll use it rebuild sweepEvent.
-			sweepTxId, sweepTx, err = s.builder.BuildSweepTx(outputsToSweep)
+			// All outputs are already spent on-chain — we just missed marking the
+			// batch swept. We rebuild the sweepEvent from the txid; the tx is never
+			// broadcast here, so a signing failure (e.g. the wallet that swept them
+			// is no longer primary or fallback) must not block reconciliation.
+			// buildAndSignSweepTx still returns the txid on signing failure, so we
+			// reconcile with the txid and store no tx bytes (an unsigned PSBT here
+			// would masquerade as a signed sweep tx in SweepTxs / the swept event).
+			sweepTxId, sweepTx, err = buildAndSignSweepTx(
+				s.builder, s.signingWallets(), outputsToSweep,
+			)
 			if err != nil {
-				return err
+				if sweepTxId == "" {
+					// the build itself failed; nothing to reconcile with
+					return err
+				}
+				log.WithError(err).Warnf(
+					"sweeper: could not sign sweep tx for already-spent batch %s, "+
+						"reconciling with txid only", commitmentTxid,
+				)
+				sweepTx = ""
 			}
 		}
 
@@ -749,7 +819,9 @@ func (s *sweeper) createCheckpointSweepTask(
 		checkpointTxid := toSweep.Txid
 		log.Debugf("sweeper: start sweeping checkpoint %s", checkpointTxid)
 
-		_, sweepTx, err := s.builder.BuildSweepTx([]ports.TxInput{toSweep})
+		_, sweepTx, err := buildAndSignSweepTx(
+			s.builder, s.signingWallets(), []ports.TxInput{toSweep},
+		)
 		if err != nil {
 			return err
 		}