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
 }
@@ -844,11 +855,11 @@ func (c *Config) walletService() error {
 
 // 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.
+// wallets belong to additional liquidity providers and are used as sweep
+// fallbacks: sweep signing is attempted with the primary first, then each
+// fallback. The primary remains the sole source of the forfeit pubkey and
+// addresses and the only wallet that funds batches. Any failure is fatal so a
+// misconfigured wallet is surfaced 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 +1019,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 +1037,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 primaryThenFallbacks(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/sweep_fallback_internal_test.go

@@ -0,0 +1,102 @@
+package application
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/arkade-os/arkd/internal/core/ports"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeWallet is a distinct, comparable ports.WalletService identity. Only its
+// pointer identity matters for the fallback-iteration tests.
+type fakeWallet struct {
+	ports.WalletService
+	name string
+}
+
+// fakeSweepBuilder implements just the two sweep methods of ports.TxBuilder that
+// buildAndSignSweepTx uses. SignSweepTx succeeds only for goodWallet, recording the
+// order in which wallets are tried.
+type fakeSweepBuilder struct {
+	ports.TxBuilder
+	buildErr   error
+	unsignedTx string
+	txid       string
+	goodWallet ports.WalletService
+	signCalls  []ports.WalletService
+}
+
+func (b *fakeSweepBuilder) BuildSweepTx(inputs []ports.TxInput) (string, string, error) {
+	if b.buildErr != nil {
+		return "", "", b.buildErr
+	}
+	return b.unsignedTx, b.txid, nil
+}
+
+func (b *fakeSweepBuilder) SignSweepTx(
+	wallet ports.WalletService, unsignedTx string,
+) (string, error) {
+	b.signCalls = append(b.signCalls, wallet)
+	if b.goodWallet != nil && wallet == b.goodWallet {
+		return "signed:" + unsignedTx, nil
+	}
+	return "", fmt.Errorf("wallet %v cannot sign", wallet)
+}
+
+func TestBuildAndSignSweepTx(t *testing.T) {
+	inputs := []ports.TxInput{{Txid: "aa", Index: 0}}
+	primary := &fakeWallet{name: "primary"}
+	fb1 := &fakeWallet{name: "fb1"}
+	fb2 := &fakeWallet{name: "fb2"}
+	wallets := []ports.WalletService{primary, fb1, fb2}
+
+	t.Run("primary signs, fallbacks not tried", func(t *testing.T) {
+		b := &fakeSweepBuilder{unsignedTx: "unsigned", txid: "txid123", goodWallet: primary}
+
+		txid, signed, err := buildAndSignSweepTx(b, wallets, inputs)
+		require.NoError(t, err)
+		require.Equal(t, "txid123", txid)
+		require.Equal(t, "signed:unsigned", signed)
+		require.Equal(t, []ports.WalletService{primary}, b.signCalls)
+	})
+
+	t.Run("falls back to a later wallet in order", func(t *testing.T) {
+		b := &fakeSweepBuilder{unsignedTx: "unsigned", txid: "txid123", goodWallet: fb2}
+
+		txid, signed, err := buildAndSignSweepTx(b, wallets, inputs)
+		require.NoError(t, err)
+		require.Equal(t, "txid123", txid)
+		require.Equal(t, "signed:unsigned", signed)
+		require.Equal(t, []ports.WalletService{primary, fb1, fb2}, b.signCalls)
+	})
+
+	t.Run("no wallet can sign returns aggregated error naming each wallet", func(t *testing.T) {
+		b := &fakeSweepBuilder{unsignedTx: "unsigned", txid: "txid123"}
+
+		txid, signed, err := buildAndSignSweepTx(b, wallets, inputs)
+		require.Error(t, err)
+		require.Empty(t, txid)
+		require.Empty(t, signed)
+		require.Contains(t, err.Error(), "no wallet could sign sweep tx txid123")
+		require.Contains(t, err.Error(), "wallet[0]")
+		require.Contains(t, err.Error(), "wallet[2]")
+		require.Len(t, b.signCalls, 3)
+	})
+
+	t.Run("no signing wallets configured", func(t *testing.T) {
+		b := &fakeSweepBuilder{unsignedTx: "unsigned", txid: "txid123"}
+
+		_, _, err := buildAndSignSweepTx(b, nil, inputs)
+		require.ErrorContains(t, err, "no signing wallets configured for sweep tx txid123")
+		require.Empty(t, b.signCalls)
+	})
+
+	t.Run("build error short-circuits before signing", func(t *testing.T) {
+		b := &fakeSweepBuilder{buildErr: fmt.Errorf("boom")}
+
+		_, _, err := buildAndSignSweepTx(b, wallets, inputs)
+		require.ErrorContains(t, err, "boom")
+		require.Empty(t, b.signCalls)
+	})
+}