fix: broadcast fan-out waits for first success instead of first result (#42)

* fix: broadcast fan-out waits for first success instead of first result

When multiple teranode endpoints are configured, the broadcast now
returns on the first successful response (ACCEPTED/SENT). Rejections
from one endpoint no longer short-circuit if another endpoint may
accept the transaction. Status is only persisted and published once
after the final result is determined.

* feat: distinguish SERVICE_ERROR from REJECTED in broadcast results

4xx responses from teranode are genuine rejections (invalid tx) and
return immediately. 5xx responses are service errors — other endpoints
may still accept the transaction. Fan-out only waits on service errors,
not rejections.

* fix: parse coinbase from block data to fix merkle proof construction

fetchBlockSubtrees was downloading the full block binary from the datahub
but only extracting subtree hashes, discarding the coinbase transaction.
The coinbase txid is needed to replace the all-FF placeholder at subtree
index 0 — without it, every hash on the left side of the merkle tree is
computed incorrectly, producing invalid proofs.

Renamed fetchBlockSubtrees to fetchBlock, now returns a blockData struct
with both SubtreeHashes and CoinbaseTxID. processBlockTransactions uses
the coinbase from block data directly instead of relying on the P2P
BlockMessage.Coinbase field, which may not always be set (e.g. during
catch-up processing).

Removed parseCoinbaseTxID as it's no longer needed.

* fix: address linter issues from broadcast and coinbase changes

* refactor: replace nolint:exhaustive with explicit status cases

* chore: bump go-jose to v4.1.4 and grpc to v1.80.0 to fix CVE-2026-34986

Author: shruggr

arcade.go

@@ -3,7 +3,6 @@ package arcade
 
 import (
 	"context"
-	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
@@ -37,6 +36,7 @@ var (
 	errChaintracksNoTip       = errors.New("chaintracks has no tip")
 	errUnexpectedStatusCode   = errors.New("unexpected status code")
 	errInvalidHashSize        = errors.New("invalid hash size")
+	errMissingCoinbaseTxID    = errors.New("block data missing coinbase txid")
 )
 
 // Config holds configuration for Arcade
@@ -335,18 +335,24 @@ func (a *Arcade) processBlockMessage(ctx context.Context, blockMsg teranode.Bloc
 }
 
 func (a *Arcade) processBlockTransactions(ctx context.Context, blockMsg teranode.BlockMessage) error { //nolint:gocyclo // complex business logic for merkle path construction
-	subtreeHashes, err := a.fetchBlockSubtreeHashes(ctx, blockMsg.DataHubURL, blockMsg.Hash)
+	block, err := a.fetchBlockData(ctx, blockMsg.DataHubURL, blockMsg.Hash)
 	if err != nil {
-		return fmt.Errorf("failed to fetch subtree hashes: %w", err)
+		return fmt.Errorf("failed to fetch block data: %w", err)
 	}
 
-	numSubtrees := len(subtreeHashes)
+	numSubtrees := len(block.SubtreeHashes)
 	if numSubtrees == 0 {
 		a.logger.Debug("block has no subtrees",
 			slog.String("hash", blockMsg.Hash))
 		return nil
 	}
 
+	if block.CoinbaseTxID == nil {
+		a.logger.Error("block data missing coinbase txid, merkle proofs will be incorrect",
+			slog.String("hash", blockMsg.Hash))
+		return fmt.Errorf("%w: %s", errMissingCoinbaseTxID, blockMsg.Hash)
+	}
+
 	a.logger.Debug("processing block transactions",
 		slog.String("hash", blockMsg.Hash),
 		slog.Int("subtrees", numSubtrees),
@@ -356,7 +362,7 @@ func (a *Arcade) processBlockTransactions(ctx context.Context, blockMsg teranode
 	totalTxsScanned := 0
 	totalMatched := 0
 
-	for subtreeIdx, subtreeHash := range subtreeHashes {
+	for subtreeIdx, subtreeHash := range block.SubtreeHashes {
 		txHashes, err := a.fetchSubtreeHashes(ctx, blockMsg.DataHubURL, subtreeHash.String())
 		if err != nil {
 			a.logger.Error("failed to fetch subtree txids",
@@ -367,12 +373,10 @@ func (a *Arcade) processBlockTransactions(ctx context.Context, blockMsg teranode
 
 		totalTxsScanned += len(txHashes)
 
-		// Subtree 0 contains a coinbase placeholder that must be replaced with the
-		// real coinbase txid. The subtree 0 root will be computed by ComputeMissingHashes.
+		// Subtree 0 contains a coinbase placeholder that must be replaced
+		// with the real coinbase txid from the block data.
 		if subtreeIdx == 0 && len(txHashes) > 0 {
-			if coinbaseTxID := a.parseCoinbaseTxID(blockMsg); coinbaseTxID != nil {
-				txHashes[0] = *coinbaseTxID
-			}
+			txHashes[0] = *block.CoinbaseTxID
 		}
 
 		tracked := a.txTracker.FilterTrackedHashes(txHashes)
@@ -395,7 +399,7 @@ func (a *Arcade) processBlockTransactions(ctx context.Context, blockMsg teranode
 			continue
 		}
 
-		a.buildMerklePathsForSubtree(ctx, blockMsg, subtreeIdx, subtreeRootLayer, subtreeHashes, txHashes, tracked)
+		a.buildMerklePathsForSubtree(ctx, blockMsg, subtreeIdx, subtreeRootLayer, block.SubtreeHashes, txHashes, tracked)
 	}
 
 	a.logger.Debug("block transaction scan complete",
@@ -407,32 +411,6 @@ func (a *Arcade) processBlockTransactions(ctx context.Context, blockMsg teranode
 	return nil
 }
 
-// parseCoinbaseTxID extracts the txid from the coinbase transaction in a block message.
-// Returns nil if the coinbase cannot be parsed.
-func (a *Arcade) parseCoinbaseTxID(blockMsg teranode.BlockMessage) *chainhash.Hash {
-	if blockMsg.Coinbase == "" {
-		return nil
-	}
-
-	coinbaseBytes, err := hex.DecodeString(blockMsg.Coinbase)
-	if err != nil {
-		a.logger.Error("failed to decode coinbase hex",
-			slog.String("blockHash", blockMsg.Hash),
-			slog.String("error", err.Error()))
-		return nil
-	}
-
-	tx, err := transaction.NewTransactionFromBytes(coinbaseBytes)
-	if err != nil {
-		a.logger.Error("failed to parse coinbase transaction",
-			slog.String("blockHash", blockMsg.Hash),
-			slog.String("error", err.Error()))
-		return nil
-	}
-
-	return tx.TxID()
-}
-
 //nolint:gocyclo // complex business logic for merkle path construction
 func (a *Arcade) buildMerklePathsForSubtree(
 	ctx context.Context,
@@ -939,11 +917,11 @@ func (a *Arcade) processRejectedTxMessage(ctx context.Context, rejectedMsg teran
 
 // HTTP fetching methods
 
-func (a *Arcade) fetchBlockSubtreeHashes(ctx context.Context, dataHubURL, blockHash string) ([]chainhash.Hash, error) {
+func (a *Arcade) fetchBlockData(ctx context.Context, dataHubURL, blockHash string) (*blockData, error) {
 	url := fmt.Sprintf("%s/block/%s", strings.TrimSuffix(dataHubURL, "/"), blockHash)
-	hashes, err := a.fetchBlockSubtrees(ctx, url)
+	data, err := a.fetchBlock(ctx, url)
 	if err == nil {
-		return hashes, nil
+		return data, nil
 	}
 
 	// Try fallback URLs
@@ -952,27 +930,32 @@ func (a *Arcade) fetchBlockSubtreeHashes(ctx context.Context, dataHubURL, blockH
 			continue // Skip if same as original
 		}
 		url = fmt.Sprintf("%s/block/%s", strings.TrimSuffix(fallbackURL, "/"), blockHash)
-		hashes, fallbackErr := a.fetchBlockSubtrees(ctx, url)
+		data, fallbackErr := a.fetchBlock(ctx, url)
 		if fallbackErr == nil {
-			return hashes, nil
+			return data, nil
 		}
 	}
 
 	return nil, err // Return original error
 }
 
-// fetchBlockSubtrees fetches a block from teranode and extracts subtree hashes from the binary format.
+// blockData holds the parsed result of a datahub block response.
+type blockData struct {
+	SubtreeHashes []chainhash.Hash
+	CoinbaseTxID  *chainhash.Hash
+}
+
+// fetchBlock fetches a block from a datahub and parses subtree hashes and the coinbase txid.
+//
 // Teranode block binary format:
 // - Block header (80 bytes)
 // - Transaction count (varint)
 // - Size in bytes (varint)
 // - Subtree count (varint)
 // - Subtree hashes (32 bytes each)
-// - Coinbase transaction
+// - Coinbase transaction (raw tx bytes)
 // - Height (varint)
-//
-//nolint:gocyclo // complex business logic for parsing merkle tree data
-func (a *Arcade) fetchBlockSubtrees(ctx context.Context, url string) ([]chainhash.Hash, error) {
+func (a *Arcade) fetchBlock(ctx context.Context, url string) (*blockData, error) { //nolint:gocyclo // sequential binary parsing
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create request: %w", err)
@@ -1029,7 +1012,18 @@ func (a *Arcade) fetchBlockSubtrees(ctx context.Context, url string) ([]chainhas
 		hashes = append(hashes, *hash)
 	}
 
-	return hashes, nil
+	result := &blockData{SubtreeHashes: hashes}
+
+	// Parse coinbase tx from the remaining stream.
+	// The stream contains: coinbase tx bytes followed by a height varint.
+	// ReadFrom reads exactly the tx and stops, leaving the height unread.
+	coinbaseTx := &transaction.Transaction{}
+	if _, err := coinbaseTx.ReadFrom(resp.Body); err != nil {
+		return nil, fmt.Errorf("failed to read coinbase tx: %w", err)
+	}
+	result.CoinbaseTxID = coinbaseTx.TxID()
+
+	return result, nil
 }
 
 func (a *Arcade) fetchSubtreeHashes(ctx context.Context, dataHubURL, subtreeHash string) ([]chainhash.Hash, error) {

go.mod

@@ -247,7 +247,7 @@ require (
 	gonum.org/v1/gonum v0.17.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260330182312-d5a96adf58d8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260330182312-d5a96adf58d8 // indirect
-	google.golang.org/grpc v1.79.3 // indirect
+	google.golang.org/grpc v1.80.0 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect

go.sum

@@ -1196,8 +1196,8 @@ google.golang.org/grpc v1.35.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAG
 google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
-google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE=
-google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ=
+google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM=
+google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=

handlers/waitfor.go

@@ -130,6 +130,7 @@ func statusReached(current, target models.Status) bool {
 	order := map[models.Status]int{
 		models.StatusUnknown:              0,
 		models.StatusReceived:             1,
+		models.StatusServiceError:         1, // transient — may still succeed
 		models.StatusSentToNetwork:        2,
 		models.StatusAcceptedByNetwork:    3,
 		models.StatusSeenOnNetwork:        4,

models/transaction.go

@@ -66,8 +66,10 @@ const (
 	StatusSeenOnNetwork = Status("SEEN_ON_NETWORK")
 	// StatusDoubleSpendAttempted indicates a double spend was attempted.
 	StatusDoubleSpendAttempted = Status("DOUBLE_SPEND_ATTEMPTED")
-	// StatusRejected indicates the transaction was rejected.
+	// StatusRejected indicates the transaction was rejected by the network (invalid tx).
 	StatusRejected = Status("REJECTED")
+	// StatusServiceError indicates a broadcast service failure (not a tx rejection).
+	StatusServiceError = Status("SERVICE_ERROR")
 	// StatusMined indicates the transaction was mined.
 	StatusMined = Status("MINED")
 	// StatusImmutable indicates the transaction is immutable.
@@ -84,6 +86,8 @@ func (s Status) DisallowedPreviousStatuses() []Status {
 		return []Status{StatusSentToNetwork, StatusAcceptedByNetwork, StatusSeenOnNetwork, StatusRejected, StatusDoubleSpendAttempted, StatusMined}
 	case StatusAcceptedByNetwork:
 		return []Status{StatusAcceptedByNetwork, StatusSeenOnNetwork, StatusRejected, StatusDoubleSpendAttempted, StatusMined}
+	case StatusServiceError:
+		return []Status{StatusAcceptedByNetwork, StatusSeenOnNetwork, StatusMined, StatusImmutable}
 	case StatusSeenOnNetwork, StatusRejected, StatusDoubleSpendAttempted, StatusMined, StatusImmutable:
 		return []Status{}
 	default: