From 059214f869629ef5b6baa8677d2f843218d98bcb Mon Sep 17 00:00:00 2001 From: Bob Smith <5396652+bitcoin-coder-bob@users.noreply.github.com> Date: Tue, 9 Jun 2026 17:05:51 -0400 Subject: [PATCH 1/3] Read UTXOs from nbxplorer events + cache wallet key paths --- .../core/application/wallet/keypath_cache.go | 45 +++ .../application/wallet/keypath_cache_test.go | 77 +++++ .../core/application/wallet/service.go | 34 ++- .../core/application/wallet/service_test.go | 206 +++++++++++++ .../core/infrastructure/nbxplorer/service.go | 107 ++++--- .../infrastructure/nbxplorer/service_test.go | 282 ++++++++++++++++++ .../core/infrastructure/nbxplorer/types.go | 32 +- pkg/arkd-wallet/core/ports/nbxplorer.go | 4 + 8 files changed, 719 insertions(+), 68 deletions(-) create mode 100644 pkg/arkd-wallet/core/application/wallet/keypath_cache.go create mode 100644 pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go create mode 100644 pkg/arkd-wallet/core/application/wallet/service_test.go create mode 100644 pkg/arkd-wallet/core/infrastructure/nbxplorer/service_test.go diff --git a/pkg/arkd-wallet/core/application/wallet/keypath_cache.go b/pkg/arkd-wallet/core/application/wallet/keypath_cache.go new file mode 100644 index 000000000..3b20fe9c3 --- /dev/null +++ b/pkg/arkd-wallet/core/application/wallet/keypath_cache.go @@ -0,0 +1,45 @@ +package wallet + +import "sync" + +// keyPathEntry records the derivation scheme and relative key path a script +// belongs to. +type keyPathEntry struct { + derivationScheme string + keyPath string +} + +// keyPathCache caches the script -> (derivation scheme, key path) mapping. +// +// The mapping is immutable: a given script always derives from the same key +// path under the same account, so cached entries never need invalidation. The +// cache lets the wallet skip the per-input NBXplorer script lookups when signing +// inputs it has already seen as UTXOs. +type keyPathCache struct { + mu sync.RWMutex + cache map[string]keyPathEntry +} + +func newKeyPathCache() *keyPathCache { + return &keyPathCache{cache: make(map[string]keyPathEntry)} +} + +// get returns the cached entry for the given script, if any. +func (c *keyPathCache) get(script string) (keyPathEntry, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + entry, ok := c.cache[script] + return entry, ok +} + +// set caches the derivation scheme and key path for the given script. Entries +// with an empty script or key path are ignored, as they cannot be used to derive +// a key later. +func (c *keyPathCache) set(script, derivationScheme, keyPath string) { + if script == "" || keyPath == "" { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.cache[script] = keyPathEntry{derivationScheme: derivationScheme, keyPath: keyPath} +} diff --git a/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go b/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go new file mode 100644 index 000000000..9a7e21c4f --- /dev/null +++ b/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go @@ -0,0 +1,77 @@ +package wallet + +import ( + "fmt" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestKeyPathCache(t *testing.T) { + t.Run("set and get", func(t *testing.T) { + c := newKeyPathCache() + + c.set("script-a", "scheme-main", "0/0") + entry, ok := c.get("script-a") + require.True(t, ok) + require.Equal(t, "scheme-main", entry.derivationScheme) + require.Equal(t, "0/0", entry.keyPath) + }) + + t.Run("get miss", func(t *testing.T) { + c := newKeyPathCache() + + _, ok := c.get("missing") + require.False(t, ok) + }) + + t.Run("empty script or key path is not cached", func(t *testing.T) { + c := newKeyPathCache() + + c.set("", "scheme", "0/0") + _, ok := c.get("") + require.False(t, ok) + + c.set("script-b", "scheme", "") + _, ok = c.get("script-b") + require.False(t, ok) + }) + + t.Run("overwrite", func(t *testing.T) { + c := newKeyPathCache() + + c.set("script-c", "scheme-main", "0/0") + c.set("script-c", "scheme-connector", "1/2") + + entry, ok := c.get("script-c") + require.True(t, ok) + require.Equal(t, "scheme-connector", entry.derivationScheme) + require.Equal(t, "1/2", entry.keyPath) + }) +} + +func TestKeyPathCache_Concurrent(t *testing.T) { + c := newKeyPathCache() + numberOfRoutines := 100 + + wg := sync.WaitGroup{} + wg.Add(numberOfRoutines) + + for i := range numberOfRoutines { + go func(i int) { + defer wg.Done() + script := fmt.Sprintf("script-%d", i) + c.set(script, "scheme", fmt.Sprintf("0/%d", i)) + _, _ = c.get(script) + }(i) + } + + wg.Wait() + + for i := range numberOfRoutines { + entry, ok := c.get(fmt.Sprintf("script-%d", i)) + require.True(t, ok) + require.Equal(t, fmt.Sprintf("0/%d", i), entry.keyPath) + } +} diff --git a/pkg/arkd-wallet/core/application/wallet/service.go b/pkg/arkd-wallet/core/application/wallet/service.go index 95c574acd..b8f7a3d79 100644 --- a/pkg/arkd-wallet/core/application/wallet/service.go +++ b/pkg/arkd-wallet/core/application/wallet/service.go @@ -54,14 +54,15 @@ type WalletOptions struct { type wallet struct { WalletOptions - locker *outpointLocker - keyMgr *keyManager - readyCh chan bool + locker *outpointLocker + keyMgr *keyManager + keyPaths *keyPathCache + readyCh chan bool } // New creates a new WalletService service func New(opts WalletOptions) application.WalletService { - return &wallet{opts, newOutpointLocker(time.Minute), nil, make(chan bool)} + return &wallet{opts, newOutpointLocker(time.Minute), nil, newKeyPathCache(), make(chan bool)} } func (w *wallet) GetReadyUpdate(ctx context.Context) <-chan bool { @@ -336,6 +337,7 @@ func (w *wallet) ListConnectorUtxos(ctx context.Context, connectorAddress string if err != nil { return nil, err } + w.cacheKeyPaths(w.keyMgr.connectorAccountDerivationScheme, connectorAccountUtxos) lockedOutpoints, err := w.locker.get(ctx) if err != nil { @@ -388,6 +390,7 @@ func (w *wallet) SelectUtxos(ctx context.Context, amount uint64, confirmedOnly b if err != nil { return nil, 0, err } + w.cacheKeyPaths(w.keyMgr.mainAccountDerivationScheme, mainAccountUtxos) lockedOutpoints, err := w.locker.get(ctx) if err != nil { @@ -832,6 +835,7 @@ func (w *wallet) withdrawAll(ctx context.Context, feeRate chainfee.SatPerKVByte, if err != nil { return nil, err } + w.cacheKeyPaths(w.keyMgr.mainAccountDerivationScheme, mainAccountUtxos) utxos = append(utxos, mainAccountUtxos...) @@ -840,6 +844,7 @@ func (w *wallet) withdrawAll(ctx context.Context, feeRate chainfee.SatPerKVByte, if err != nil { return nil, err } + w.cacheKeyPaths(w.keyMgr.connectorAccountDerivationScheme, connectorAccountUtxos) utxos = append(utxos, connectorAccountUtxos...) } @@ -985,6 +990,12 @@ func (w *wallet) getPrivateKeyFromScript(ctx context.Context, scriptPubKey strin return nil, ErrWalletLocked } + // A cache hit lets us derive the key without any NBXplorer lookup. The cache + // is populated whenever we list UTXOs, which already carry their key path. + if entry, ok := w.keyPaths.get(scriptPubKey); ok { + return w.keyMgr.deriveKey(entry.derivationScheme, entry.keyPath) + } + accountsDerivationSchemes := []string{ w.keyMgr.mainAccountDerivationScheme, w.keyMgr.connectorAccountDerivationScheme, @@ -995,18 +1006,33 @@ func (w *wallet) getPrivateKeyFromScript(ctx context.Context, scriptPubKey strin if err != nil { continue } + // a script tracked under a scheme always has a key path; if it is empty + // the script does not belong to this account, so try the next scheme. + if scriptPubKeyDetails.KeyPath == "" { + continue + } + w.keyPaths.set(scriptPubKey, derivationScheme, scriptPubKeyDetails.KeyPath) return w.keyMgr.deriveKey(derivationScheme, scriptPubKeyDetails.KeyPath) } return nil, nil } +// cacheKeyPaths records the script -> key path mapping for the given UTXOs so +// that later signing of these scripts can skip the per-input NBXplorer lookup. +func (w *wallet) cacheKeyPaths(derivationScheme string, utxos []ports.Utxo) { + for _, utxo := range utxos { + w.keyPaths.set(utxo.Script, derivationScheme, utxo.KeyPath) + } +} + func (w *wallet) getBalance(ctx context.Context, derivationScheme string) (uint64, uint64, error) { utxos, err := w.Nbxplorer.GetUtxos(ctx, derivationScheme) if err != nil { return 0, 0, err } + w.cacheKeyPaths(derivationScheme, utxos) lockedOutpoints, err := w.locker.get(ctx) if err != nil { diff --git a/pkg/arkd-wallet/core/application/wallet/service_test.go b/pkg/arkd-wallet/core/application/wallet/service_test.go new file mode 100644 index 000000000..8bd7b96b2 --- /dev/null +++ b/pkg/arkd-wallet/core/application/wallet/service_test.go @@ -0,0 +1,206 @@ +package wallet + +import ( + "context" + "fmt" + "testing" + + "github.com/arkade-os/arkd/pkg/arkd-wallet/core/ports" + "github.com/btcsuite/btcd/chaincfg" + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/lnwallet/chainfee" + "github.com/stretchr/testify/require" +) + +// fakeNbxplorer is a minimal ports.Nbxplorer used to count script lookups. +type fakeNbxplorer struct { + getScriptCalls int + scriptDetails func(scheme, script string) (*ports.ScriptPubKeyDetails, error) +} + +var _ ports.Nbxplorer = (*fakeNbxplorer)(nil) + +func (f *fakeNbxplorer) GetScriptPubKeyDetails( + _ context.Context, scheme, script string, +) (*ports.ScriptPubKeyDetails, error) { + f.getScriptCalls++ + if f.scriptDetails != nil { + return f.scriptDetails(scheme, script) + } + return nil, fmt.Errorf("script not found") +} + +func (f *fakeNbxplorer) GetBitcoinStatus(_ context.Context) (*ports.BitcoinStatus, error) { + return nil, nil +} + +func (f *fakeNbxplorer) GetTransaction( + _ context.Context, _ string, +) (*ports.TransactionDetails, error) { + return nil, nil +} + +func (f *fakeNbxplorer) ScanUtxoSet( + _ context.Context, _ string, _ int, +) <-chan ports.ScanUtxoSetProgress { + return nil +} + +func (f *fakeNbxplorer) Track(_ context.Context, _ string) error { return nil } + +func (f *fakeNbxplorer) GetUtxos(_ context.Context, _ string) ([]ports.Utxo, error) { + return nil, nil +} + +func (f *fakeNbxplorer) GetNewUnusedAddress( + _ context.Context, _ string, _ bool, _ int, +) (string, error) { + return "", nil +} + +func (f *fakeNbxplorer) EstimateFeeRate(_ context.Context) (chainfee.SatPerKVByte, error) { + return 0, nil +} + +func (f *fakeNbxplorer) BroadcastTransaction(_ context.Context, _ ...string) (string, error) { + return "", nil +} + +func (f *fakeNbxplorer) RescanUtxos(_ context.Context, _ []wire.OutPoint) error { return nil } + +func (f *fakeNbxplorer) IsSpent(_ context.Context, _ wire.OutPoint) (bool, error) { + return false, nil +} + +func (f *fakeNbxplorer) WatchAddresses(_ context.Context, _ ...string) error { return nil } +func (f *fakeNbxplorer) UnwatchAddresses(_ context.Context, _ ...string) error { return nil } + +func (f *fakeNbxplorer) GetAddressNotifications( + _ context.Context, +) (<-chan []ports.Utxo, error) { + return nil, nil +} + +func (f *fakeNbxplorer) Close() error { return nil } + +func newTestWallet(t *testing.T, nbx ports.Nbxplorer) (*wallet, *keyManager) { + t.Helper() + seed := make([]byte, 32) + for i := range seed { + seed[i] = 0x01 + } + km, err := newKeyManager(seed, &chaincfg.RegressionNetParams) + require.NoError(t, err) + + w := &wallet{ + WalletOptions: WalletOptions{Nbxplorer: nbx, Network: "regtest"}, + keyMgr: km, + keyPaths: newKeyPathCache(), + } + return w, km +} + +func TestGetPrivateKeyFromScript(t *testing.T) { + const script = "5120deadbeef" + + t.Run("locked wallet", func(t *testing.T) { + w := &wallet{keyPaths: newKeyPathCache()} + _, err := w.getPrivateKeyFromScript(context.Background(), script) + require.ErrorIs(t, err, ErrWalletLocked) + }) + + t.Run("cache hit avoids nbxplorer lookup", func(t *testing.T) { + fake := &fakeNbxplorer{} + w, km := newTestWallet(t, fake) + w.keyPaths.set(script, km.mainAccountDerivationScheme, "0/0") + + key, err := w.getPrivateKeyFromScript(context.Background(), script) + require.NoError(t, err) + require.NotNil(t, key) + require.Equal(t, 0, fake.getScriptCalls) + }) + + t.Run("cache miss falls back then backfills", func(t *testing.T) { + fake := &fakeNbxplorer{} + w, km := newTestWallet(t, fake) + fake.scriptDetails = func(scheme, _ string) (*ports.ScriptPubKeyDetails, error) { + if scheme == km.mainAccountDerivationScheme { + return &ports.ScriptPubKeyDetails{KeyPath: "0/1"}, nil + } + return nil, fmt.Errorf("not found") + } + + key, err := w.getPrivateKeyFromScript(context.Background(), script) + require.NoError(t, err) + require.NotNil(t, key) + require.Equal(t, 1, fake.getScriptCalls) + + // the second call must be served from the cache, no extra lookup. + key2, err := w.getPrivateKeyFromScript(context.Background(), script) + require.NoError(t, err) + require.NotNil(t, key2) + require.Equal(t, 1, fake.getScriptCalls) + }) + + t.Run("connector fallback performs two lookups", func(t *testing.T) { + fake := &fakeNbxplorer{} + w, km := newTestWallet(t, fake) + fake.scriptDetails = func(scheme, _ string) (*ports.ScriptPubKeyDetails, error) { + if scheme == km.connectorAccountDerivationScheme { + return &ports.ScriptPubKeyDetails{KeyPath: "0/2"}, nil + } + return nil, fmt.Errorf("not found") + } + + key, err := w.getPrivateKeyFromScript(context.Background(), script) + require.NoError(t, err) + require.NotNil(t, key) + require.Equal(t, 2, fake.getScriptCalls) + }) + + t.Run("empty key path from main scheme falls back to connector", func(t *testing.T) { + fake := &fakeNbxplorer{} + w, km := newTestWallet(t, fake) + fake.scriptDetails = func(scheme, _ string) (*ports.ScriptPubKeyDetails, error) { + if scheme == km.connectorAccountDerivationScheme { + return &ports.ScriptPubKeyDetails{KeyPath: "0/3"}, nil + } + // main scheme answers without error but with an empty key path + return &ports.ScriptPubKeyDetails{KeyPath: ""}, nil + } + + key, err := w.getPrivateKeyFromScript(context.Background(), script) + require.NoError(t, err) + require.NotNil(t, key) + require.Equal(t, 2, fake.getScriptCalls) + }) + + t.Run("unknown script returns nil key after trying both schemes", func(t *testing.T) { + fake := &fakeNbxplorer{} + w, _ := newTestWallet(t, fake) + + key, err := w.getPrivateKeyFromScript(context.Background(), script) + require.NoError(t, err) + require.Nil(t, key) + require.Equal(t, 2, fake.getScriptCalls) + }) +} + +func TestCacheKeyPaths(t *testing.T) { + w := &wallet{keyPaths: newKeyPathCache()} + utxos := []ports.Utxo{ + {Script: "s1", KeyPath: "0/0"}, + {Script: "s2", KeyPath: "0/1"}, + {Script: "s3", KeyPath: ""}, // no key path: must not be cached + } + + w.cacheKeyPaths("scheme-main", utxos) + + entry, ok := w.keyPaths.get("s1") + require.True(t, ok) + require.Equal(t, "scheme-main", entry.derivationScheme) + require.Equal(t, "0/0", entry.keyPath) + + _, ok = w.keyPaths.get("s3") + require.False(t, ok) +} diff --git a/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go b/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go index 10970d0b0..a20afb76d 100644 --- a/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go +++ b/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go @@ -679,28 +679,19 @@ func (n *nbxplorer) GetAddressNotifications(ctx context.Context) (<-chan []ports continue } - var event event - if err := json.Unmarshal(message, &event); err != nil { + // The matched outputs are carried in the event payload itself, so + // the new UTXOs can be built without any additional NBXplorer call. + newUtxos, err := utxosFromTransactionEvent(message, n.groupID) + if err != nil { + log.Errorf("failed to parse transaction event: %s", err) continue } - if event.Type == "newtransaction" { - var newTxEvent newTransactionEvent - if eventDataBytes, err := json.Marshal(event.Data); err == nil { - if err := json.Unmarshal(eventDataBytes, &newTxEvent); err == nil { - newUtxos, err := n.searchNewUTXOs(ctx, newTxEvent.TransactionData.TransactionHash) - if err != nil { - continue - } - - if len(newUtxos) > 0 { - select { - case notificationsChan <- newUtxos: - case <-ctx.Done(): - return - } - } - } + if len(newUtxos) > 0 { + select { + case notificationsChan <- newUtxos: + case <-ctx.Done(): + return } } } @@ -809,52 +800,55 @@ func (n *nbxplorer) connectWebSocket(ctx context.Context) error { return nil } -// searchNewUTXOs rescans UTXOs for a specific group and returns only UTXOs from the specified transaction hash -func (n *nbxplorer) searchNewUTXOs(ctx context.Context, txHash string) ([]ports.Utxo, error) { - if txHash == "" { - return nil, fmt.Errorf("transaction hash is required") - } - - cryptoCode := btcCryptoCode - endpoint := fmt.Sprintf("/v1/cryptos/%s/groups/%s/utxos", cryptoCode, url.PathEscape(n.groupID)) +// groupTrackedSource returns the tracked-source identifier NBXplorer uses for a +// watch-address group. Addresses added via the group addresses endpoint report +// matches under this identifier. +func groupTrackedSource(groupID string) string { + return fmt.Sprintf("GROUP:%s", groupID) +} - data, err := n.makeRequest(ctx, "GET", endpoint, nil) - if err != nil { - return nil, fmt.Errorf("failed to get group UTXOs: %w", err) +// utxosFromTransactionEvent parses a websocket message and, if it is a +// "newtransaction" event for the watch-address group, returns the UTXOs created +// by that transaction. The matched outputs are read directly from the event +// payload, so no additional NBXplorer call is needed. Events for any other +// tracked source (e.g. the wallet's own derivation schemes) are ignored. +func utxosFromTransactionEvent(message []byte, groupID string) ([]ports.Utxo, error) { + var evt event + if err := json.Unmarshal(message, &evt); err != nil { + return nil, fmt.Errorf("failed to unmarshal event: %w", err) } - var resp utxosResponse - if err := json.Unmarshal(data, &resp); err != nil { - return nil, fmt.Errorf("failed to unmarshal UTXO changes: %w", err) + if evt.Type != "newtransaction" { + return nil, nil } - utxos := make([]ports.Utxo, 0) - - for _, u := range resp.Confirmed.UtxOs { - if u.TransactionHash != txHash { - continue - } - - utxo, err := castUtxo(u) - if err != nil { - log.Errorf("failed to cast UTXO: %s", err) - continue - } - - utxos = append(utxos, utxo) + var newTxEvent newTransactionEvent + if err := json.Unmarshal(evt.Data, &newTxEvent); err != nil { + return nil, fmt.Errorf("failed to unmarshal new transaction event: %w", err) } - for _, u := range resp.Unconfirmed.UtxOs { - if u.TransactionHash != txHash { - continue - } + if newTxEvent.TrackedSource != groupTrackedSource(groupID) { + return nil, nil + } - utxo, err := castUtxo(u) - if err != nil { - continue - } + hash, err := chainhash.NewHashFromStr(newTxEvent.TransactionData.TransactionHash) + if err != nil { + return nil, fmt.Errorf("failed to parse transaction hash: %w", err) + } - utxos = append(utxos, utxo) + utxos := make([]ports.Utxo, 0, len(newTxEvent.Outputs)) + for _, out := range newTxEvent.Outputs { + utxos = append(utxos, ports.Utxo{ + OutPoint: wire.OutPoint{ + Hash: *hash, + Index: out.Index, + }, + Value: out.Value, + Script: out.ScriptPubKey, + Address: out.Address, + Confirmations: newTxEvent.TransactionData.Confirmations, + KeyPath: out.KeyPath, + }) } return utxos, nil @@ -903,5 +897,6 @@ func castUtxo(u utxoResponse) (ports.Utxo, error) { Script: u.ScriptPubKey, Address: u.Address, Confirmations: u.Confirmations, + KeyPath: u.KeyPath, }, nil } diff --git a/pkg/arkd-wallet/core/infrastructure/nbxplorer/service_test.go b/pkg/arkd-wallet/core/infrastructure/nbxplorer/service_test.go new file mode 100644 index 000000000..17358a9b9 --- /dev/null +++ b/pkg/arkd-wallet/core/infrastructure/nbxplorer/service_test.go @@ -0,0 +1,282 @@ +package nbxplorer + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/arkade-os/arkd/pkg/arkd-wallet/core/ports" + "github.com/stretchr/testify/require" +) + +const ( + testGroupID = "test-group-id" + testTxID = "0000000000000000000000000000000000000000000000000000000000000001" +) + +func testScript() string { + return "5120" + strings.Repeat("11", 32) +} + +// newTransactionEventJSON builds a raw websocket "newtransaction" event message +// with the given tracked source, transaction id, confirmation count and outputs +// JSON array. +func newTransactionEventJSON( + trackedSource, txid string, confirmations uint32, outputs string, +) []byte { + return []byte(fmt.Sprintf( + `{"type":"newtransaction","eventId":1,"data":{`+ + `"trackedSource":%q,"cryptoCode":"BTC",`+ + `"transactionData":{"transactionHash":%q,"confirmations":%d,"timestamp":1700000000},`+ + `"outputs":%s}}`, + trackedSource, txid, confirmations, outputs, + )) +} + +func output(script, address string, index, value uint64, keyPath string) string { + kp := "" + if keyPath != "" { + kp = fmt.Sprintf(`"keyPath":%q,`, keyPath) + } + return fmt.Sprintf( + `{%s"scriptPubKey":%q,"index":%d,"keyIndex":0,"value":%d,"address":%q}`, + kp, script, index, value, address, + ) +} + +func TestGroupTrackedSource(t *testing.T) { + require.Equal(t, "GROUP:abc", groupTrackedSource("abc")) +} + +func TestUtxosFromTransactionEvent(t *testing.T) { + script := testScript() + groupTS := groupTrackedSource(testGroupID) + + t.Run("group match single output", func(t *testing.T) { + msg := newTransactionEventJSON( + groupTS, testTxID, 3, "["+output(script, "bcrt1paddr", 1, 100000, "")+"]", + ) + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Len(t, utxos, 1) + + u := utxos[0] + require.Equal(t, testTxID, u.OutPoint.Hash.String()) + require.Equal(t, uint32(1), u.OutPoint.Index) + require.Equal(t, uint64(100000), u.Value) + require.Equal(t, script, u.Script) + require.Equal(t, "bcrt1paddr", u.Address) + require.Equal(t, uint32(3), u.Confirmations) + require.Empty(t, u.KeyPath) + }) + + t.Run("group match empty outputs", func(t *testing.T) { + msg := newTransactionEventJSON(groupTS, testTxID, 0, "[]") + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Empty(t, utxos) + }) + + t.Run("group match multiple outputs", func(t *testing.T) { + outputs := "[" + + output(script, "addr0", 0, 1000, "") + "," + + output(script, "addr1", 1, 2000, "") + "]" + msg := newTransactionEventJSON(groupTS, testTxID, 0, outputs) + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Len(t, utxos, 2) + require.Equal(t, uint32(0), utxos[0].OutPoint.Index) + require.Equal(t, uint32(1), utxos[1].OutPoint.Index) + require.Equal(t, uint64(1000), utxos[0].Value) + require.Equal(t, uint64(2000), utxos[1].Value) + }) + + t.Run("keypath carried through", func(t *testing.T) { + msg := newTransactionEventJSON( + groupTS, testTxID, 0, "["+output(script, "addr", 0, 1000, "0/5")+"]", + ) + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Len(t, utxos, 1) + require.Equal(t, "0/5", utxos[0].KeyPath) + }) + + t.Run("different group is ignored", func(t *testing.T) { + msg := newTransactionEventJSON( + "GROUP:some-other-group", testTxID, 0, + "["+output(script, "addr", 0, 1000, "")+"]", + ) + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Empty(t, utxos) + }) + + t.Run("derivation scheme tracked source is ignored", func(t *testing.T) { + msg := newTransactionEventJSON( + "DERIVATIONSCHEME:xpub-[taproot]", testTxID, 0, + "["+output(script, "addr", 0, 1000, "")+"]", + ) + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Empty(t, utxos) + }) + + t.Run("non-newtransaction event is ignored", func(t *testing.T) { + msg := []byte(`{"type":"newblock","data":{"height":100,"hash":"abc"}}`) + + utxos, err := utxosFromTransactionEvent(msg, testGroupID) + require.NoError(t, err) + require.Empty(t, utxos) + }) + + t.Run("invalid json returns error", func(t *testing.T) { + _, err := utxosFromTransactionEvent([]byte("not json"), testGroupID) + require.Error(t, err) + }) + + t.Run("invalid transaction hash returns error", func(t *testing.T) { + msg := newTransactionEventJSON( + groupTS, "not-a-hash", 0, "["+output(script, "addr", 0, 1000, "")+"]", + ) + + _, err := utxosFromTransactionEvent(msg, testGroupID) + require.Error(t, err) + }) +} + +func TestCastUtxoCarriesKeyPath(t *testing.T) { + u := utxoResponse{ + TransactionHash: testTxID, + Index: 2, + ScriptPubKey: testScript(), + Address: "bcrt1paddr", + Value: 5000, + KeyPath: "1/9", + Confirmations: 3, + } + + utxo, err := castUtxo(u) + require.NoError(t, err) + require.Equal(t, "1/9", utxo.KeyPath) + require.Equal(t, uint32(2), utxo.OutPoint.Index) + require.Equal(t, uint64(5000), utxo.Value) + require.Equal(t, uint32(3), utxo.Confirmations) +} + +// fetchAndFilterGroupUtxos replicates the previous (pre-optimization) behavior: +// on every event it fetched the whole group UTXO set over HTTP and filtered it +// client-side by transaction hash. It is kept here only as a benchmark baseline. +func fetchAndFilterGroupUtxos(client *http.Client, url, txHash string) ([]ports.Utxo, error) { + resp, err := client.Get(url) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + data, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + var r utxosResponse + if err := json.Unmarshal(data, &r); err != nil { + return nil, err + } + + utxos := make([]ports.Utxo, 0) + for _, u := range r.Confirmed.UtxOs { + if u.TransactionHash != txHash { + continue + } + utxo, err := castUtxo(u) + if err != nil { + continue + } + utxos = append(utxos, utxo) + } + for _, u := range r.Unconfirmed.UtxOs { + if u.TransactionHash != txHash { + continue + } + utxo, err := castUtxo(u) + if err != nil { + continue + } + utxos = append(utxos, utxo) + } + return utxos, nil +} + +func buildGroupUtxosBody(n int, matchTxHash string) []byte { + var resp utxosResponse + resp.Confirmed.UtxOs = make([]utxoResponse, 0, n) + for i := 0; i < n; i++ { + // offset by 2 so no filler hash collides with matchTxHash (which is 0x..01) + resp.Confirmed.UtxOs = append(resp.Confirmed.UtxOs, utxoResponse{ + TransactionHash: fmt.Sprintf("%064x", i+2), + Index: 0, + ScriptPubKey: testScript(), + Address: "bcrt1paddr", + Value: 1000, + Confirmations: 1, + }) + } + if n > 0 { + resp.Confirmed.UtxOs[n-1].TransactionHash = matchTxHash + } + b, _ := json.Marshal(resp) + return b +} + +// BenchmarkNotificationProcessing compares the per-event work of the new +// event-payload parsing against the old approach of fetching and filtering the +// whole group UTXO set. The old approach scales with the group size; the new one +// does not depend on it and performs no network round trip. +func BenchmarkNotificationProcessing(b *testing.B) { + eventMsg := newTransactionEventJSON( + groupTrackedSource(testGroupID), testTxID, 0, + "["+output(testScript(), "bcrt1paddr", 1, 100000, "")+"]", + ) + + b.Run("event-parse", func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + utxos, err := utxosFromTransactionEvent(eventMsg, testGroupID) + if err != nil || len(utxos) != 1 { + b.Fatalf("unexpected result: %v %v", utxos, err) + } + } + }) + + for _, n := range []int{10, 100, 1000, 10000} { + body := buildGroupUtxosBody(n, testTxID) + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(body) + }, + )) + client := srv.Client() + + b.Run(fmt.Sprintf("http-fetch-group-size-%d", n), func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + utxos, err := fetchAndFilterGroupUtxos(client, srv.URL, testTxID) + if err != nil || len(utxos) != 1 { + b.Fatalf("unexpected result: %v %v", utxos, err) + } + } + }) + + srv.Close() + } +} diff --git a/pkg/arkd-wallet/core/infrastructure/nbxplorer/types.go b/pkg/arkd-wallet/core/infrastructure/nbxplorer/types.go index 77faa6de1..828aceab8 100644 --- a/pkg/arkd-wallet/core/infrastructure/nbxplorer/types.go +++ b/pkg/arkd-wallet/core/infrastructure/nbxplorer/types.go @@ -1,5 +1,7 @@ package nbxplorer +import "encoding/json" + type bitcoinStatusResponse struct { BitcoinStatus struct { Blocks uint32 `json:"blocks"` @@ -111,19 +113,33 @@ type rpcError struct { } type event struct { - EventID int `json:"eventId"` - Type string `json:"type"` - Data interface{} `json:"data"` + EventID int `json:"eventId"` + Type string `json:"type"` + Data json.RawMessage `json:"data"` +} + +// matchedOutput is an output of a transaction that matched a tracked source. +// It is embedded in newtransaction events, carrying everything needed to build +// a UTXO without an additional NBXplorer call. +type matchedOutput struct { + KeyPath string `json:"keyPath,omitempty"` + ScriptPubKey string `json:"scriptPubKey"` + Index uint32 `json:"index"` + KeyIndex uint32 `json:"keyIndex"` + Feature string `json:"feature,omitempty"` + Value uint64 `json:"value"` + Address string `json:"address"` } type newTransactionEvent struct { - BlockID string `json:"blockId,omitempty"` - TrackedSource string `json:"trackedSource"` - DerivationStrategy string `json:"derivationStrategy"` - CryptoCode string `json:"cryptoCode"` + BlockID string `json:"blockId,omitempty"` + TrackedSource string `json:"trackedSource"` + DerivationStrategy string `json:"derivationStrategy"` + CryptoCode string `json:"cryptoCode"` + Outputs []matchedOutput `json:"outputs"` TransactionData struct { TransactionHash string `json:"transactionHash"` - Confirmations int `json:"confirmations"` + Confirmations uint32 `json:"confirmations"` Height *int `json:"height,omitempty"` Timestamp int64 `json:"timestamp"` } `json:"transactionData"` diff --git a/pkg/arkd-wallet/core/ports/nbxplorer.go b/pkg/arkd-wallet/core/ports/nbxplorer.go index 2db70998e..ef3a74bc6 100644 --- a/pkg/arkd-wallet/core/ports/nbxplorer.go +++ b/pkg/arkd-wallet/core/ports/nbxplorer.go @@ -28,6 +28,10 @@ type Utxo struct { Script string Address string Confirmations uint32 + // KeyPath is the derivation path of the script relative to its account + // derivation scheme (e.g. "0/5"). It may be empty for scripts that are not + // derived from a tracked derivation scheme (e.g. watch-only addresses). + KeyPath string } type ScriptPubKeyDetails struct { From 923414ac479f5b45bd787957adbd65d8496d25fa Mon Sep 17 00:00:00 2001 From: Bob Smith <5396652+bitcoin-coder-bob@users.noreply.github.com> Date: Wed, 10 Jun 2026 12:37:45 -0400 Subject: [PATCH 2/3] Bound key path cache with LRU + named wallet struct fields --- .../core/application/wallet/keypath_cache.go | 43 +++++++++++-------- .../application/wallet/keypath_cache_test.go | 20 +++++++++ .../core/application/wallet/service.go | 7 ++- pkg/arkd-wallet/go.mod | 1 + 4 files changed, 52 insertions(+), 19 deletions(-) diff --git a/pkg/arkd-wallet/core/application/wallet/keypath_cache.go b/pkg/arkd-wallet/core/application/wallet/keypath_cache.go index 3b20fe9c3..b1f82c902 100644 --- a/pkg/arkd-wallet/core/application/wallet/keypath_cache.go +++ b/pkg/arkd-wallet/core/application/wallet/keypath_cache.go @@ -1,6 +1,15 @@ package wallet -import "sync" +import lru "github.com/hashicorp/golang-lru/v2" + +// keyPathCacheSize bounds the number of cached script -> key path entries. +// +// A signing operation only needs the scripts of the inputs being signed, which +// are the most recently listed UTXOs, so evicting older entries is always safe: +// a miss simply falls back to the NBXplorer lookup. The bound keeps memory flat +// on a long-running server whose wallet churns through many addresses (e.g. the +// connector account across many rounds). +const keyPathCacheSize = 50_000 // keyPathEntry records the derivation scheme and relative key path a script // belongs to. @@ -9,37 +18,35 @@ type keyPathEntry struct { keyPath string } -// keyPathCache caches the script -> (derivation scheme, key path) mapping. +// keyPathCache is a bounded, concurrency-safe cache of the script -> +// (derivation scheme, key path) mapping. // // The mapping is immutable: a given script always derives from the same key -// path under the same account, so cached entries never need invalidation. The -// cache lets the wallet skip the per-input NBXplorer script lookups when signing -// inputs it has already seen as UTXOs. +// path under the same account, so cached entries are never stale. The cache lets +// the wallet skip the per-input NBXplorer script lookups when signing inputs it +// has already seen as UTXOs. type keyPathCache struct { - mu sync.RWMutex - cache map[string]keyPathEntry + cache *lru.Cache[string, keyPathEntry] } func newKeyPathCache() *keyPathCache { - return &keyPathCache{cache: make(map[string]keyPathEntry)} + // lru.New only errors on a non-positive size, which is a positive constant + // here, so the error cannot occur. + cache, _ := lru.New[string, keyPathEntry](keyPathCacheSize) + return &keyPathCache{cache: cache} } // get returns the cached entry for the given script, if any. func (c *keyPathCache) get(script string) (keyPathEntry, bool) { - c.mu.RLock() - defer c.mu.RUnlock() - entry, ok := c.cache[script] - return entry, ok + return c.cache.Get(script) } -// set caches the derivation scheme and key path for the given script. Entries -// with an empty script or key path are ignored, as they cannot be used to derive -// a key later. +// set caches the derivation scheme and key path for the given script, evicting +// the least-recently-used entry if the cache is full. Entries with an empty +// script or key path are ignored, as they cannot be used to derive a key later. func (c *keyPathCache) set(script, derivationScheme, keyPath string) { if script == "" || keyPath == "" { return } - c.mu.Lock() - defer c.mu.Unlock() - c.cache[script] = keyPathEntry{derivationScheme: derivationScheme, keyPath: keyPath} + c.cache.Add(script, keyPathEntry{derivationScheme: derivationScheme, keyPath: keyPath}) } diff --git a/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go b/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go index 9a7e21c4f..2019eb800 100644 --- a/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go +++ b/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go @@ -51,6 +51,26 @@ func TestKeyPathCache(t *testing.T) { }) } +func TestKeyPathCache_Bounded(t *testing.T) { + c := newKeyPathCache() + + total := keyPathCacheSize + 1000 + for i := range total { + c.set(fmt.Sprintf("script-%d", i), "scheme", fmt.Sprintf("0/%d", i)) + } + + // the cache never grows past its bound + require.Equal(t, keyPathCacheSize, c.cache.Len()) + + // the oldest entries are evicted, the most recent are retained + _, ok := c.get("script-0") + require.False(t, ok) + + entry, ok := c.get(fmt.Sprintf("script-%d", total-1)) + require.True(t, ok) + require.Equal(t, fmt.Sprintf("0/%d", total-1), entry.keyPath) +} + func TestKeyPathCache_Concurrent(t *testing.T) { c := newKeyPathCache() numberOfRoutines := 100 diff --git a/pkg/arkd-wallet/core/application/wallet/service.go b/pkg/arkd-wallet/core/application/wallet/service.go index b8f7a3d79..d44077be9 100644 --- a/pkg/arkd-wallet/core/application/wallet/service.go +++ b/pkg/arkd-wallet/core/application/wallet/service.go @@ -62,7 +62,12 @@ type wallet struct { // New creates a new WalletService service func New(opts WalletOptions) application.WalletService { - return &wallet{opts, newOutpointLocker(time.Minute), nil, newKeyPathCache(), make(chan bool)} + return &wallet{ + WalletOptions: opts, + locker: newOutpointLocker(time.Minute), + keyPaths: newKeyPathCache(), + readyCh: make(chan bool), + } } func (w *wallet) GetReadyUpdate(ctx context.Context) <-chan bool { diff --git a/pkg/arkd-wallet/go.mod b/pkg/arkd-wallet/go.mod index 2d9907ee5..9f3a6d4cb 100644 --- a/pkg/arkd-wallet/go.mod +++ b/pkg/arkd-wallet/go.mod @@ -16,6 +16,7 @@ require ( github.com/google/uuid v1.6.0 github.com/grafana/pyroscope-go v1.2.7 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 + github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/lightningnetwork/lnd v0.18.2-beta github.com/sirupsen/logrus v1.9.3 github.com/spf13/viper v1.20.1 From 7138bb9bcd972ef1ddd0406c97d63328e896d497 Mon Sep 17 00:00:00 2001 From: Bob Smith <5396652+bitcoin-coder-bob@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:18:15 -0400 Subject: [PATCH 3/3] arkd-wallet: group the key path cache tests, warn on an unparseable event Regroup only for the tests. The parse failure of a notification frame is benign for the wallet and was previously swallowed, so it logs at warn. --- .../application/wallet/keypath_cache_test.go | 78 +++++++++---------- .../core/infrastructure/nbxplorer/service.go | 2 +- 2 files changed, 40 insertions(+), 40 deletions(-) diff --git a/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go b/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go index 2019eb800..76558a728 100644 --- a/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go +++ b/pkg/arkd-wallet/core/application/wallet/keypath_cache_test.go @@ -49,49 +49,49 @@ func TestKeyPathCache(t *testing.T) { require.Equal(t, "scheme-connector", entry.derivationScheme) require.Equal(t, "1/2", entry.keyPath) }) -} - -func TestKeyPathCache_Bounded(t *testing.T) { - c := newKeyPathCache() - - total := keyPathCacheSize + 1000 - for i := range total { - c.set(fmt.Sprintf("script-%d", i), "scheme", fmt.Sprintf("0/%d", i)) - } - - // the cache never grows past its bound - require.Equal(t, keyPathCacheSize, c.cache.Len()) - - // the oldest entries are evicted, the most recent are retained - _, ok := c.get("script-0") - require.False(t, ok) - - entry, ok := c.get(fmt.Sprintf("script-%d", total-1)) - require.True(t, ok) - require.Equal(t, fmt.Sprintf("0/%d", total-1), entry.keyPath) -} -func TestKeyPathCache_Concurrent(t *testing.T) { - c := newKeyPathCache() - numberOfRoutines := 100 + t.Run("bounded", func(t *testing.T) { + c := newKeyPathCache() - wg := sync.WaitGroup{} - wg.Add(numberOfRoutines) + total := keyPathCacheSize + 1000 + for i := range total { + c.set(fmt.Sprintf("script-%d", i), "scheme", fmt.Sprintf("0/%d", i)) + } - for i := range numberOfRoutines { - go func(i int) { - defer wg.Done() - script := fmt.Sprintf("script-%d", i) - c.set(script, "scheme", fmt.Sprintf("0/%d", i)) - _, _ = c.get(script) - }(i) - } + // the cache never grows past its bound + require.Equal(t, keyPathCacheSize, c.cache.Len()) - wg.Wait() + // the oldest entries are evicted, the most recent are retained + _, ok := c.get("script-0") + require.False(t, ok) - for i := range numberOfRoutines { - entry, ok := c.get(fmt.Sprintf("script-%d", i)) + entry, ok := c.get(fmt.Sprintf("script-%d", total-1)) require.True(t, ok) - require.Equal(t, fmt.Sprintf("0/%d", i), entry.keyPath) - } + require.Equal(t, fmt.Sprintf("0/%d", total-1), entry.keyPath) + }) + + t.Run("concurrent", func(t *testing.T) { + c := newKeyPathCache() + numberOfRoutines := 100 + + wg := sync.WaitGroup{} + wg.Add(numberOfRoutines) + + for i := range numberOfRoutines { + go func(i int) { + defer wg.Done() + script := fmt.Sprintf("script-%d", i) + c.set(script, "scheme", fmt.Sprintf("0/%d", i)) + _, _ = c.get(script) + }(i) + } + + wg.Wait() + + for i := range numberOfRoutines { + entry, ok := c.get(fmt.Sprintf("script-%d", i)) + require.True(t, ok) + require.Equal(t, fmt.Sprintf("0/%d", i), entry.keyPath) + } + }) } diff --git a/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go b/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go index 3101576ec..901cb0748 100644 --- a/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go +++ b/pkg/arkd-wallet/core/infrastructure/nbxplorer/service.go @@ -699,7 +699,7 @@ func (n *nbxplorer) GetAddressNotifications(ctx context.Context) (<-chan []ports // the new UTXOs can be built without any additional NBXplorer call. newUtxos, err := utxosFromTransactionEvent(message, n.groupID) if err != nil { - log.Errorf("failed to parse transaction event: %s", err) + log.Warnf("failed to parse transaction event: %s", err) continue }