Merge remote-tracking branch 'upstream/master'

Author: actions-user

Makefile

@@ -27,8 +27,23 @@ test: .bin-image
 test-integration: .bin-image
 	docker run -t --rm -e PACKAGER=$(PACKAGER) -e BB_BUILD_ENV=$(BB_BUILD_ENV) -e GITCOMMIT=$(GITCOMMIT) $(BB_RPC_ENV) -v "$(CURDIR):/src" --network="host" $(BIN_IMAGE) make test-integration ARGS="$(ARGS)"
 
+# `make test-e2e [coin] [url]` — e.g. `make test-e2e bitcoin https://btc.trezor.io`.
+# Extra goals after test-e2e are consumed as positional args (coin, base URL)
+# and turned into OPENAPI_COINS / BB_DEV_API_URL_HTTP_<coin> for the test runner.
+ifeq (test-e2e,$(firstword $(MAKECMDGOALS)))
+  E2E_ARGS := $(wordlist 2,$(words $(MAKECMDGOALS)),$(MAKECMDGOALS))
+  ifneq ($(E2E_ARGS),)
+    # no-op catch-all so make does not try to build the positional args as targets
+%:
+	@:
+  endif
+endif
+
 test-e2e:
 	@if [ ! -x tests/openapi/node_modules/.bin/redocly ]; then npm ci --prefix tests/openapi --prefer-offline --no-audit --no-fund; fi
+	@coin="$(word 1,$(E2E_ARGS))"; url="$(word 2,$(E2E_ARGS))"; \
+	if [ -n "$$coin" ]; then export OPENAPI_COINS="$$coin"; fi; \
+	if [ -n "$$url" ]; then export "BB_DEV_API_URL_HTTP_$$(printf %s "$$coin" | tr - _)=$$url"; fi; \
 	contrib/tests/run-openapi-tests.sh
 
 test-connectivity: .bin-image

api/worker.go

@@ -2045,7 +2045,7 @@ func (w *Worker) waitForBackendSync() {
 	}
 }
 
-func (w *Worker) getAddrDescUtxo(addrDesc bchain.AddressDescriptor, ba *db.AddrBalance, onlyConfirmed bool, onlyMempool bool) (Utxos, error) {
+func (w *Worker) getAddrDescUtxo(addrDesc bchain.AddressDescriptor, ba *db.AddrBalance, onlyConfirmed bool, onlyMempool bool, knownBestHeight *uint32) (Utxos, error) {
 	w.waitForBackendSync()
 	var err error
 	utxos := make(Utxos, 0, 8)
@@ -2115,11 +2115,16 @@ func (w *Worker) getAddrDescUtxo(addrDesc bchain.AddressDescriptor, ba *db.AddrB
 		}
 		// ba can be nil if the address is only in mempool!
 		if ba != nil && len(ba.Utxos) > 0 {
-			b, _, err := w.db.GetBestBlock()
-			if err != nil {
-				return nil, err
+			var bestheight int
+			if knownBestHeight != nil {
+				bestheight = int(*knownBestHeight)
+			} else {
+				b, _, err := w.db.GetBestBlock()
+				if err != nil {
+					return nil, err
+				}
+				bestheight = int(b)
 			}
-			bestheight := int(b)
 			var checksum big.Int
 			checksum.Set(&ba.BalanceSat)
 			// go backwards to get the newest first
@@ -2175,7 +2180,7 @@ func (w *Worker) GetAddressUtxo(address string, onlyConfirmed bool) (Utxos, erro
 	if err != nil {
 		return nil, NewAPIError(fmt.Sprintf("Invalid address '%v', %v", address, err), true)
 	}
-	r, err := w.getAddrDescUtxo(addrDesc, nil, onlyConfirmed, false)
+	r, err := w.getAddrDescUtxo(addrDesc, nil, onlyConfirmed, false, nil)
 	if err != nil {
 		return nil, err
 	}

api/xpub.go

@@ -69,6 +69,7 @@ type xpubData struct {
 	txCountEstimate uint32
 	sentSat         big.Int
 	balanceSat      big.Int
+	mergedTxids     xpubTxids
 	addresses       [][]xpubAddress
 }
 
@@ -197,10 +198,32 @@ func (w *Worker) xpubGetAddressTxids(addrDesc bchain.AddressDescriptor, mempool
 	return txs, complete, nil
 }
 
-func (w *Worker) xpubCheckAndLoadTxids(ad *xpubAddress, filter *AddressFilter, maxHeight uint32, maxResults int) error {
+func isUnfilteredXpubTxidFilter(filter *AddressFilter) bool {
+	return filter == nil || filter.FromHeight == 0 && filter.ToHeight == 0 && filter.Vout == AddressFilterVoutOff
+}
+
+func mergeXpubTxids(data *xpubData) xpubTxids {
+	txcMap := make(map[string]struct{}, data.txCountEstimate)
+	txc := make(xpubTxids, 0, data.txCountEstimate)
+	for _, da := range data.addresses {
+		for i := range da {
+			for _, txid := range da[i].txids {
+				if _, foundTx := txcMap[txid.txid]; foundTx {
+					continue
+				}
+				txcMap[txid.txid] = struct{}{}
+				txc = append(txc, txid)
+			}
+		}
+	}
+	sort.Stable(txc)
+	return txc
+}
+
+func (w *Worker) xpubCheckAndLoadTxids(ad *xpubAddress, maxHeight uint32, maxResults int) (bool, error) {
 	// skip if not used
 	if ad.balance == nil {
-		return nil
+		return false, nil
 	}
 	// if completely loaded, check if there are not some new txs and load if necessary
 	if ad.complete {
@@ -214,14 +237,14 @@ func (w *Worker) xpubCheckAndLoadTxids(ad *xpubAddress, filter *AddressFilter, m
 					glog.Warning("xpubCheckAndLoadTxids inconsistency ", ad.addrDesc, ", ad.txs=", ad.txs, ", ad.balance.Txs=", ad.balance.Txs)
 				}
 			}
-			return err
+			return err == nil, err
 		}
-		return nil
+		return false, nil
 	}
 	// load all txids to get paging correctly
 	newTxids, complete, err := w.xpubGetAddressTxids(ad.addrDesc, false, 0, maxHeight, maxInt)
 	if err != nil {
-		return err
+		return false, err
 	}
 	ad.txids = newTxids
 	ad.complete = complete
@@ -232,7 +255,7 @@ func (w *Worker) xpubCheckAndLoadTxids(ad *xpubAddress, filter *AddressFilter, m
 			glog.Warning("xpubCheckAndLoadTxids inconsistency ", ad.addrDesc, ", ad.txs=", ad.txs, ", ad.balance.Txs=", ad.balance.Txs)
 		}
 	}
-	return nil
+	return true, nil
 }
 
 func (w *Worker) xpubDerivedAddressBalance(data *xpubData, ad *xpubAddress) (bool, error) {
@@ -420,6 +443,7 @@ func (w *Worker) getXpubData(xd *bchain.XpubDescriptor, page int, txsOnPage int,
 			data.balanceSat = *new(big.Int)
 			data.sentSat = *new(big.Int)
 			data.txCountEstimate = 0
+			data.mergedTxids = nil
 			var minDerivedIndex int
 			totalDerived := 0
 			for i, change := range xd.ChangeIndexes {
@@ -431,13 +455,22 @@ func (w *Worker) getXpubData(xd *bchain.XpubDescriptor, page int, txsOnPage int,
 			}
 		}
 		if option >= AccountDetailsTxidHistory {
+			txidsChanged := false
 			for _, da := range data.addresses {
 				for i := range da {
-					if err = w.xpubCheckAndLoadTxids(&da[i], filter, bestheight, (page+1)*txsOnPage); err != nil {
+					changed := false
+					if changed, err = w.xpubCheckAndLoadTxids(&da[i], bestheight, (page+1)*txsOnPage); err != nil {
 						return nil, 0, inCache, err
 					}
+					txidsChanged = txidsChanged || changed
 				}
 			}
+			if txidsChanged {
+				data.mergedTxids = nil
+			}
+			if isUnfilteredXpubTxidFilter(filter) && data.mergedTxids == nil {
+				data.mergedTxids = mergeXpubTxids(&data)
+			}
 		}
 	}
 	data.accessed = time.Now().Unix()
@@ -564,30 +597,35 @@ func (w *Worker) GetXpubAddress(xpub string, page int, txsOnPage int, option Acc
 		}
 	}
 	if option >= AccountDetailsTxidHistory {
-		txcMap := make(map[string]bool)
-		txc = make(xpubTxids, 0, 32)
-		for _, da := range data.addresses {
-			for i := range da {
-				ad := &da[i]
-				for _, txid := range ad.txids {
-					added, foundTx := txcMap[txid.txid]
-					// count txs regardless of filter but only once
-					if !foundTx {
-						txCount++
-					}
-					// add tx only once
-					if !added {
-						add := txidFilter == nil || txidFilter(&txid, ad)
-						txcMap[txid.txid] = add
-						if add {
-							txc = append(txc, txid)
+		if txidFilter == nil {
+			// Shared with xpubData cache; do not mutate in this request path.
+			txc = data.mergedTxids
+			if txc == nil {
+				txc = mergeXpubTxids(data)
+			}
+			txCount = len(txc)
+		} else {
+			txcMap := make(map[string]bool)
+			txc = make(xpubTxids, 0, 32)
+			for _, da := range data.addresses {
+				for i := range da {
+					ad := &da[i]
+					for _, txid := range ad.txids {
+						added := txcMap[txid.txid]
+						// add tx only once
+						if !added {
+							add := txidFilter(&txid, ad)
+							txcMap[txid.txid] = add
+							if add {
+								txc = append(txc, txid)
+							}
 						}
 					}
 				}
 			}
+			sort.Stable(txc)
+			txCount = len(txcMap)
 		}
-		sort.Stable(txc)
-		txCount = len(txcMap)
 		totalResults := txCount
 		if filtered {
 			totalResults = -1
@@ -692,7 +730,7 @@ func (w *Worker) GetXpubUtxo(xpub string, onlyConfirmed bool, gap int) (Utxos, e
 	if err != nil {
 		return nil, err
 	}
-	data, _, inCache, err := w.getXpubData(xd, 0, 1, AccountDetailsBasic, &AddressFilter{
+	data, bestheight, inCache, err := w.getXpubData(xd, 0, 1, AccountDetailsBasic, &AddressFilter{
 		Vout:          AddressFilterVoutOff,
 		OnlyConfirmed: onlyConfirmed,
 	}, gap)
@@ -710,7 +748,7 @@ func (w *Worker) GetXpubUtxo(xpub string, onlyConfirmed bool, gap int) (Utxos, e
 				}
 				onlyMempool = true
 			}
-			utxos, err := w.getAddrDescUtxo(ad.addrDesc, ad.balance, onlyConfirmed, onlyMempool)
+			utxos, err := w.getAddrDescUtxo(ad.addrDesc, ad.balance, onlyConfirmed, onlyMempool, &bestheight)
 			if err != nil {
 				return nil, err
 			}

api/xpub_test.go

@@ -52,3 +52,48 @@ func TestTrimXpubCacheItemsLocked(t *testing.T) {
 		t.Fatal("second oldest cache entry was not evicted")
 	}
 }
+
+func TestMergeXpubTxidsDeduplicatesAndSorts(t *testing.T) {
+	data := &xpubData{
+		txCountEstimate: 4,
+		addresses: [][]xpubAddress{
+			{
+				{txids: xpubTxids{
+					{txid: "duplicate", height: 5, inputOutput: txOutput},
+					{txid: "newest", height: 7, inputOutput: txOutput},
+				}},
+			},
+			{
+				{txids: xpubTxids{
+					{txid: "duplicate", height: 5, inputOutput: txInput},
+					{txid: "same-height-input", height: 5, inputOutput: txInput},
+				}},
+			},
+		},
+	}
+
+	txids := mergeXpubTxids(data)
+	got := make([]string, len(txids))
+	for i := range txids {
+		got[i] = txids[i].txid
+	}
+	want := []string{"newest", "same-height-input", "duplicate"}
+	if fmt.Sprint(got) != fmt.Sprint(want) {
+		t.Fatalf("mergeXpubTxids order = %v, want %v", got, want)
+	}
+	if txids[2].inputOutput != txOutput {
+		t.Fatal("mergeXpubTxids did not preserve the first duplicate occurrence")
+	}
+}
+
+func TestIsUnfilteredXpubTxidFilter(t *testing.T) {
+	if !isUnfilteredXpubTxidFilter(&AddressFilter{Vout: AddressFilterVoutOff}) {
+		t.Fatal("default xpub txid filter should be unfiltered")
+	}
+	if isUnfilteredXpubTxidFilter(&AddressFilter{Vout: AddressFilterVoutInputs}) {
+		t.Fatal("input filter should not be treated as unfiltered")
+	}
+	if isUnfilteredXpubTxidFilter(&AddressFilter{Vout: AddressFilterVoutOff, FromHeight: 1}) {
+		t.Fatal("height filter should not be treated as unfiltered")
+	}
+}

server/websocket.go

@@ -159,6 +159,7 @@ func NewWebsocketServer(db *db.RocksDB, chain bchain.BlockChain, mempool bchain.
 	s.upgrader = &websocket.Upgrader{
 		ReadBufferSize:    1024 * 32,
 		WriteBufferSize:   1024 * 32,
+		WriteBufferPool:   &sync.Pool{},
 		CheckOrigin:       s.checkOrigin,
 		EnableCompression: true,
 	}
@@ -1797,53 +1798,36 @@ func (s *WebsocketServer) sendOnNewTxAddr(stringAddressDescriptor string, tx *ap
 
 func (s *WebsocketServer) getNewTxSubscriptions(vins []bchain.MempoolVin, vouts []bchain.Vout, tokenTransfers bchain.TokenTransfers, internalTransfers []bchain.EthereumInternalTransfer, newBlockTxsOnly bool) map[string]struct{} {
 	// check if there is any subscription in inputs, outputs and transfers
-	s.addressSubscriptionsLock.Lock()
-	defer s.addressSubscriptionsLock.Unlock()
-	subscribed := make(map[string]struct{})
-	hasSubscription := func(sad string) bool {
-		as, ok := s.addressSubscriptions[sad]
-		if !ok || len(as) == 0 {
-			return false
-		}
-		if !newBlockTxsOnly {
-			return true
-		}
-		for _, details := range as {
-			if details.publishNewBlockTxs {
-				return true
-			}
+	candidates := make(map[string]struct{})
+	addAddrDesc := func(addrDesc bchain.AddressDescriptor) {
+		if len(addrDesc) > 0 {
+			candidates[string(addrDesc)] = struct{}{}
 		}
-		return false
 	}
 	processAddress := func(address string) {
 		if addrDesc, err := s.chainParser.GetAddrDescFromAddress(address); err == nil && len(addrDesc) > 0 {
-			sad := string(addrDesc)
-			if hasSubscription(sad) {
-				subscribed[sad] = struct{}{}
-			}
+			addAddrDesc(addrDesc)
 		}
 	}
 	processVout := func(vout bchain.Vout) {
 		if addrDesc, err := s.chainParser.GetAddrDescFromVout(&vout); err == nil && len(addrDesc) > 0 {
-			sad := string(addrDesc)
-			if hasSubscription(sad) {
-				subscribed[sad] = struct{}{}
-			}
+			addAddrDesc(addrDesc)
 		}
 	}
 	for i := range vins {
 		if sad := string(vins[i].AddrDesc); len(sad) > 0 {
-			if hasSubscription(sad) {
-				subscribed[sad] = struct{}{}
-			}
-		} else if s.chainParser.GetChainType() == bchain.ChainBitcoinType {
-			vout := int(vins[i].Vout)
-			if vout >= 0 && vout < len(vouts) {
-				processVout(vouts[vins[i].Vout])
-			}
-		} else if s.chainParser.GetChainType() == bchain.ChainEthereumType {
-			if len(vins[i].Addresses) > 0 {
-				processAddress(vins[i].Addresses[0])
+			candidates[sad] = struct{}{}
+		} else {
+			switch s.chainParser.GetChainType() {
+			case bchain.ChainBitcoinType:
+				vout := int(vins[i].Vout)
+				if vout >= 0 && vout < len(vouts) {
+					processVout(vouts[vout])
+				}
+			case bchain.ChainEthereumType:
+				if len(vins[i].Addresses) > 0 {
+					processAddress(vins[i].Addresses[0])
+				}
 			}
 		}
 	}
@@ -1858,6 +1842,26 @@ func (s *WebsocketServer) getNewTxSubscriptions(vins []bchain.MempoolVin, vouts
 		processAddress(internalTransfers[i].From)
 		processAddress(internalTransfers[i].To)
 	}
+
+	subscribed := make(map[string]struct{})
+	s.addressSubscriptionsLock.Lock()
+	defer s.addressSubscriptionsLock.Unlock()
+	for sad := range candidates {
+		as, ok := s.addressSubscriptions[sad]
+		if !ok || len(as) == 0 {
+			continue
+		}
+		if !newBlockTxsOnly {
+			subscribed[sad] = struct{}{}
+			continue
+		}
+		for _, details := range as {
+			if details.publishNewBlockTxs {
+				subscribed[sad] = struct{}{}
+				break
+			}
+		}
+	}
 	return subscribed
 }