Merge pull request #1530 from lightninglabs/1520-skip-send-transfer-anchor-broadcast

Add support for skipping anchor transaction broadcast during asset transfer

Author: ffranr

itest/mint_fund_seal_test.go

@@ -464,7 +464,7 @@ func testMintFundSealAssets(t *harnessTest) {
 
 	// Finalize and publish the transfer anchor TX.
 	signedPkt := FinalizePacket(t.t, aliceLnd.RPC, transferPkt)
-	logResp := LogAndPublish(
+	logResp := PublishAndLogTransfer(
 		t.t, aliceTapd, signedPkt, bobVpsbt, passiveVpsbts,
 		commitResp,
 	)

itest/multi_send_test.go

@@ -374,7 +374,7 @@ func testAnchorMultipleVirtualTransactions(t *harnessTest) {
 
 	btcPacket = signPacket(t.t, aliceLnd, btcPacket)
 	btcPacket = FinalizePacket(t.t, aliceLnd.RPC, btcPacket)
-	sendResp = LogAndPublish(
+	sendResp = PublishAndLogTransfer(
 		t.t, aliceTapd, btcPacket, activeAssets, passiveAssets,
 		commitResp,
 	)

itest/multisig.go

@@ -334,7 +334,7 @@ func MultiSigTest(t *testing.T, ctx context.Context, aliceTapd,
 	// transaction.
 	signedPkt := FinalizePacket(t, bobLnd, btcWithdrawPkt)
 
-	logResp := LogAndPublish(
+	logResp := PublishAndLogTransfer(
 		t, bobTapd, signedPkt, finalizedWithdrawPackets, nil,
 		commitResp,
 	)
@@ -490,10 +490,57 @@ func FinalizePacket(t *testing.T, lnd *rpc.HarnessRPC,
 	return signedPacket
 }
 
-func LogAndPublish(t *testing.T, tapd commands.RpcClientsBundle,
+// PublishAndLogTransferOption defines a functional option for
+// PublishAndLogTransfer.
+type PublishAndLogTransferOption func(*publishAndLogTransferOptions)
+
+// publishAndLogTransferOptions contains the options for PublishAndLogTransfer.
+type publishAndLogTransferOptions struct {
+	// skipAnchorTxBroadcast indicates whether to skip the broadcast of
+	// the anchor transaction.
+	skipAnchorTxBroadcast bool
+
+	// label is the label to use for the transfer.
+	label string
+}
+
+// defaultPublishAndLogTransferOptions returns the default options for
+// PublishAndLogTransfer.
+func defaultPublishAndLogTransferOptions() *publishAndLogTransferOptions {
+	return &publishAndLogTransferOptions{}
+}
+
+// withSkipAnchorTxBroadcast is an option for PublishAndLogTransfer that
+// indicates whether to skip the broadcast of the anchor transaction.
+func withSkipAnchorTxBroadcast() PublishAndLogTransferOption {
+	return func(opts *publishAndLogTransferOptions) {
+		opts.skipAnchorTxBroadcast = true
+	}
+}
+
+// withLabel is an option for PublishAndLogTransfer that sets the label for
+// the transfer.
+func withLabel(label string) PublishAndLogTransferOption {
+	return func(opts *publishAndLogTransferOptions) {
+		opts.label = label
+	}
+}
+
+// PublishAndLogTransfer is a helper function that invokes the
+// PublishAndLogTransfer RPC endpoint on the specified tapd node. This endpoint
+// performs a pre-anchored transfer.
+func PublishAndLogTransfer(t *testing.T, tapd commands.RpcClientsBundle,
 	btcPkt *psbt.Packet, activeAssets []*tappsbt.VPacket,
 	passiveAssets []*tappsbt.VPacket,
-	commitResp *wrpc.CommitVirtualPsbtsResponse) *taprpc.SendAssetResponse {
+	commitResp *wrpc.CommitVirtualPsbtsResponse,
+	opts ...PublishAndLogTransferOption) *taprpc.SendAssetResponse {
+
+	t.Helper()
+
+	options := defaultPublishAndLogTransferOptions()
+	for _, opt := range opts {
+		opt(options)
+	}
 
 	ctxb := context.Background()
 	ctxt, cancel := context.WithTimeout(ctxb, defaultWaitTimeout)
@@ -504,11 +551,13 @@ func LogAndPublish(t *testing.T, tapd commands.RpcClientsBundle,
 	require.NoError(t, err)
 
 	request := &wrpc.PublishAndLogRequest{
-		AnchorPsbt:        buf.Bytes(),
-		VirtualPsbts:      make([][]byte, len(activeAssets)),
-		PassiveAssetPsbts: make([][]byte, len(passiveAssets)),
-		ChangeOutputIndex: commitResp.ChangeOutputIndex,
-		LndLockedUtxos:    commitResp.LndLockedUtxos,
+		AnchorPsbt:            buf.Bytes(),
+		VirtualPsbts:          make([][]byte, len(activeAssets)),
+		PassiveAssetPsbts:     make([][]byte, len(passiveAssets)),
+		ChangeOutputIndex:     commitResp.ChangeOutputIndex,
+		LndLockedUtxos:        commitResp.LndLockedUtxos,
+		SkipAnchorTxBroadcast: options.skipAnchorTxBroadcast,
+		Label:                 options.label,
 	}
 
 	for idx := range activeAssets {

itest/psbt_test.go

@@ -7,12 +7,15 @@ import (
 	"fmt"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/btcsuite/btcd/btcec/v2/schnorr"
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/btcutil/hdkeychain"
 	"github.com/btcsuite/btcd/btcutil/psbt"
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcwallet/waddrmgr"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightninglabs/taproot-assets/address"
@@ -1124,7 +1127,7 @@ func testPsbtInteractiveAltLeafAnchoring(t *harnessTest) {
 
 	commitPacket = signPacket(t.t, senderLnd, commitPacket)
 	commitPacket = FinalizePacket(t.t, senderLnd.RPC, commitPacket)
-	publishResp := LogAndPublish(
+	publishResp := PublishAndLogTransfer(
 		t.t, sender, commitPacket, []*tappsbt.VPacket{activePacket},
 		[]*tappsbt.VPacket{passivePacket}, commitResp,
 	)
@@ -2438,7 +2441,7 @@ func testPsbtTrustlessSwap(t *harnessTest) {
 	signedPkt := finalizePacket(t.t, lndBob, finalPsbt)
 	require.True(t.t, signedPkt.IsComplete())
 
-	logResp := logAndPublish(
+	logResp := PublishAndLogTransfer(
 		t.t, alice, signedPkt, []*tappsbt.VPacket{bobVPsbt}, nil, resp,
 	)
 	t.Logf("Logged transaction: %v", toJSON(t.t, logResp))
@@ -2611,18 +2614,80 @@ func testPsbtExternalCommit(t *harnessTest) {
 
 	btcPacket = signPacket(t.t, aliceLnd, btcPacket)
 	btcPacket = FinalizePacket(t.t, aliceLnd.RPC, btcPacket)
-	sendResp := LogAndPublish(
+
+	transferLabel := "itest-psbt-external-commit"
+
+	// Subscribe to the send event stream so we can verify the sender's
+	// state transitions during this test.
+	ctx, streamCancel := context.WithCancel(ctxb)
+	stream, err := aliceTapd.SubscribeSendEvents(
+		ctx, &taprpc.SubscribeSendEventsRequest{
+			FilterLabel: transferLabel,
+		},
+	)
+	require.NoError(t.t, err)
+	sendEvents := &EventSubscription[*taprpc.SendEvent]{
+		ClientEventStream: stream,
+		Cancel:            streamCancel,
+	}
+
+	// Execute the transfer but skip anchor transaction broadcast. This
+	// simulates a packaging workflow where broadcasting is handled
+	// externally.
+	sendResp := PublishAndLogTransfer(
 		t.t, aliceTapd, btcPacket, activeAssets, passiveAssets,
-		commitResp,
+		commitResp, withSkipAnchorTxBroadcast(),
+		withLabel(transferLabel),
 	)
 
+	// Assert that the state machine transitions directly to waiting for
+	// tx confirmation, skipping the broadcast state.
+	require.Eventually(t.t, func() bool {
+		isMatchingState := func(msg *taprpc.SendEvent) bool {
+			lastState := tapfreighter.SendStateStorePreBroadcast
+			nextState := tapfreighter.SendStateWaitTxConf
+
+			return msg.SendState == lastState.String() &&
+				msg.NextSendState == nextState.String()
+		}
+
+		for {
+			msg, err := sendEvents.Recv()
+			if err != nil {
+				return false
+			}
+
+			return isMatchingState(msg)
+		}
+	}, defaultWaitTimeout, time.Second)
+
+	// Unmarshal the anchor transaction.
+	var anchorTx wire.MsgTx
+	reader := bytes.NewReader(sendResp.Transfer.AnchorTx)
+	err = anchorTx.Deserialize(reader)
+	require.NoError(t.t, err)
+
+	// Ensure that the anchor PSBT matches the returned anchor
+	// transaction.
+	require.Equal(t.t, anchorTx.TxHash(), btcPacket.UnsignedTx.TxHash())
+
+	// Assert that the anchor transaction is not in the mempool.
+	miner := t.lndHarness.Miner()
+	miner.AssertTxnsNotInMempool([]chainhash.Hash{
+		anchorTx.TxHash(),
+	})
+
+	// Add the anchor transaction to the mempool and mine.
+	miner.MineBlockWithTx(&anchorTx)
+
+	// Assert that the transfer has the correct number of outputs.
 	expectedAmounts := []uint64{
 		targetAsset.Amount - assetsToSend, assetsToSend,
 	}
-	ConfirmAndAssertOutboundTransferWithOutputs(
+	AssertAssetOutboundTransferWithOutputs(
 		t.t, t.lndHarness.Miner().Client, aliceTapd,
-		sendResp, targetAssetGenesis.AssetId, expectedAmounts,
-		0, 1, len(expectedAmounts),
+		sendResp.Transfer, targetAssetGenesis.AssetId, expectedAmounts,
+		0, 1, len(expectedAmounts), false,
 	)
 
 	// And now the event should be completed on both sides.
@@ -3273,7 +3338,7 @@ func testPsbtRelativeLockTimeSendProofFail(t *harnessTest) {
 		Cancel:            streamCancel,
 	}
 
-	LogAndPublish(t.t, bob, btcPacket, vPackets, nil, commitResp)
+	PublishAndLogTransfer(t.t, bob, btcPacket, vPackets, nil, commitResp)
 
 	MineBlocks(t.t, t.lndHarness.Miner().Client, 1, 1)
 
@@ -3424,45 +3489,6 @@ func finalizePacket(t *testing.T, lnd *node.HarnessNode,
 	return signedPacket
 }
 
-func logAndPublish(t *testing.T, tapd *tapdHarness, btcPkt *psbt.Packet,
-	activeAssets []*tappsbt.VPacket, passiveAssets []*tappsbt.VPacket,
-	commitResp *wrpc.CommitVirtualPsbtsResponse) *taprpc.SendAssetResponse {
-
-	ctxb := context.Background()
-	ctxt, cancel := context.WithTimeout(ctxb, defaultWaitTimeout)
-	defer cancel()
-
-	var buf bytes.Buffer
-	err := btcPkt.Serialize(&buf)
-	require.NoError(t, err)
-
-	request := &wrpc.PublishAndLogRequest{
-		AnchorPsbt:        buf.Bytes(),
-		VirtualPsbts:      make([][]byte, len(activeAssets)),
-		PassiveAssetPsbts: make([][]byte, len(passiveAssets)),
-		ChangeOutputIndex: commitResp.ChangeOutputIndex,
-		LndLockedUtxos:    commitResp.LndLockedUtxos,
-	}
-
-	for idx := range activeAssets {
-		request.VirtualPsbts[idx], err = tappsbt.Encode(
-			activeAssets[idx],
-		)
-		require.NoError(t, err)
-	}
-	for idx := range passiveAssets {
-		request.PassiveAssetPsbts[idx], err = tappsbt.Encode(
-			passiveAssets[idx],
-		)
-		require.NoError(t, err)
-	}
-
-	resp, err := tapd.PublishAndLogTransfer(ctxt, request)
-	require.NoError(t, err)
-
-	return resp
-}
-
 // getAddressBip32Derivation returns the PSBT BIP-0032 derivation info of an
 // address.
 func getAddressBip32Derivation(t testing.TB, addr string,

rpcserver.go

@@ -2994,6 +2994,7 @@ func (r *rpcServer) PublishAndLogTransfer(ctx context.Context,
 	resp, err := r.cfg.ChainPorter.RequestShipment(
 		tapfreighter.NewPreAnchoredParcel(
 			activePackets, passivePackets, anchorTx,
+			req.SkipAnchorTxBroadcast, req.Label,
 		),
 	)
 	if err != nil {
@@ -3713,6 +3714,19 @@ func marshalOutboundParcel(
 		}
 	})
 
+	// Serialize the anchor transaction if it exists.
+	var anchorTxBytes []byte
+	if parcel.AnchorTx != nil {
+		var b bytes.Buffer
+		err := parcel.AnchorTx.Serialize(&b)
+		if err != nil {
+			return nil, fmt.Errorf("unable to serialize anchor "+
+				"tx: %w", err)
+		}
+
+		anchorTxBytes = b.Bytes()
+	}
+
 	return &taprpc.AssetTransfer{
 		TransferTimestamp:   parcel.TransferTime.Unix(),
 		AnchorTxHash:        anchorTxHash[:],
@@ -3723,6 +3737,7 @@ func marshalOutboundParcel(
 		Inputs:              rpcInputs,
 		Outputs:             rpcOutputs,
 		Label:               parcel.Label,
+		AnchorTx:            anchorTxBytes,
 	}, nil
 }
 
@@ -4242,6 +4257,7 @@ func marshalSendEvent(event fn.Event) (*taprpc.SendEvent, error) {
 		VirtualPackets:        make([][]byte, len(e.VirtualPackets)),
 		PassiveVirtualPackets: make([][]byte, len(e.PassivePackets)),
 		TransferLabel:         e.TransferLabel,
+		NextSendState:         e.NextSendState.String(),
 	}
 
 	if e.Error != nil {

tapchannel/aux_closer.go

@@ -645,7 +645,7 @@ func shipChannelTxn(txSender tapfreighter.Porter, chanTx *wire.MsgTx,
 		FinalTx:   chanTx,
 	}
 	preSignedParcel := tapfreighter.NewPreAnchoredParcel(
-		vPkts, nil, closeAnchor,
+		vPkts, nil, closeAnchor, false, "",
 	)
 	_, err = txSender.RequestShipment(preSignedParcel)
 	if err != nil {

tapchannel/aux_funding_controller.go

@@ -1392,7 +1392,7 @@ func (f *FundingController) completeChannelFunding(ctx context.Context,
 		FinalTx:   signedFundingTx,
 	}
 	preSignedParcel := tapfreighter.NewPreAnchoredParcel(
-		activePkts, passivePkts, anchorTx,
+		activePkts, passivePkts, anchorTx, false, "",
 	)
 	_, err = f.cfg.TxSender.RequestShipment(preSignedParcel)
 	if err != nil {

tapdb/assets_store.go

@@ -2442,10 +2442,11 @@ func (a *AssetStore) LogPendingParcel(ctx context.Context,
 		// outputs will reference. We'll insert this next, so we can
 		// use its ID.
 		transferID, err := q.InsertAssetTransfer(ctx, NewAssetTransfer{
-			HeightHint:       int32(spend.AnchorTxHeightHint),
-			AnchorTxid:       newAnchorTXID[:],
-			TransferTimeUnix: spend.TransferTime,
-			Label:            sqlStr(spend.Label),
+			HeightHint:            int32(spend.AnchorTxHeightHint),
+			AnchorTxid:            newAnchorTXID[:],
+			TransferTimeUnix:      spend.TransferTime,
+			Label:                 sqlStr(spend.Label),
+			SkipAnchorTxBroadcast: spend.SkipAnchorTxBroadcast,
 		})
 		if err != nil {
 			return fmt.Errorf("unable to insert asset transfer: "+
@@ -3545,15 +3546,17 @@ func (a *AssetStore) QueryParcels(ctx context.Context,
 				)
 			}
 
+			// nolint:lll
 			parcel := &tapfreighter.OutboundParcel{
-				AnchorTx:           anchorTx,
-				AnchorTxHeightHint: uint32(dbT.HeightHint),
-				AnchorTxBlockHash:  anchorTxBlockHash,
-				TransferTime:       dbT.TransferTimeUnix.UTC(),
-				ChainFees:          dbAnchorTx.ChainFees,
-				Inputs:             inputs,
-				Outputs:            outputs,
-				Label:              dbT.Label.String,
+				AnchorTx:              anchorTx,
+				AnchorTxHeightHint:    uint32(dbT.HeightHint),
+				AnchorTxBlockHash:     anchorTxBlockHash,
+				TransferTime:          dbT.TransferTimeUnix.UTC(),
+				ChainFees:             dbAnchorTx.ChainFees,
+				Inputs:                inputs,
+				Outputs:               outputs,
+				Label:                 dbT.Label.String,
+				SkipAnchorTxBroadcast: dbT.SkipAnchorTxBroadcast,
 			}
 
 			// Set the block height if the anchor is marked as

tapdb/migrations.go

@@ -24,7 +24,7 @@ const (
 	// daemon.
 	//
 	// NOTE: This MUST be updated when a new migration is added.
-	LatestMigrationVersion = 34
+	LatestMigrationVersion = 35
 )
 
 // DatabaseBackend is an interface that contains all methods our different

tapdb/sqlc/migrations/000035_asset_transfers_skip_anchor_broadcast.down.sql

@@ -0,0 +1,2 @@
+-- Remove the skip_anchor_tx_broadcast flag from asset_transfers table.
+ALTER TABLE asset_transfers DROP COLUMN skip_anchor_tx_broadcast;