tapdb+tapfreighter: hold lease for a year when publishing TX

To avoid an issue that we sometimes ran into in lnd where a UTXO isn't
locked/leased after a spend transaction is published, we now lease any
spent outputs for a whole year, right before we publish the spend TX.

tapdb/assets_store.go

@@ -1811,9 +1811,13 @@ func (a *AssetStore) queryCommitments(ctx context.Context,
 
 // LogPendingParcel marks an outbound parcel as pending on disk. This commits
 // the set of changes to disk (the pending inputs and outputs) but doesn't mark
-// the batched spend as being finalized.
+// the batched spend as being finalized. The final lease owner and expiry are
+// the lease parameters that are set on the input UTXOs, since we assume the
+// parcel will be broadcast after this call. So we'll want to lock the input
+// UTXOs for forever, which means the expiry should be far in the future.
 func (a *AssetStore) LogPendingParcel(ctx context.Context,
-	spend *tapfreighter.OutboundParcel) error {
+	spend *tapfreighter.OutboundParcel, finalLeaseOwner [32]byte,
+	finalLeaseExpiry time.Time) error {
 
 	// Before we enter the DB transaction below, we'll use this space to
 	// encode a few values outside the transaction closure.
@@ -1855,6 +1859,7 @@ func (a *AssetStore) LogPendingParcel(ctx context.Context,
 		for idx := range spend.Inputs {
 			err := insertAssetTransferInput(
 				ctx, q, transferID, spend.Inputs[idx],
+				finalLeaseOwner, finalLeaseExpiry,
 			)
 			if err != nil {
 				return fmt.Errorf("unable to insert asset "+
@@ -1880,7 +1885,8 @@ func (a *AssetStore) LogPendingParcel(ctx context.Context,
 
 // insertAssetTransferInput inserts a new asset transfer input into the DB.
 func insertAssetTransferInput(ctx context.Context, q ActiveAssetsStore,
-	transferID int32, input tapfreighter.TransferInput) error {
+	transferID int32, input tapfreighter.TransferInput,
+	finalLeaseOwner [32]byte, finalLeaseExpiry time.Time) error {
 
 	anchorPointBytes, err := encodeOutpoint(input.OutPoint)
 	if err != nil {
@@ -1898,7 +1904,17 @@ func insertAssetTransferInput(ctx context.Context, q ActiveAssetsStore,
 		return fmt.Errorf("unable to insert transfer input: %w", err)
 	}
 
-	return nil
+	// From this point onward, we'll attempt to broadcast the anchor
+	// transaction, even if we restart. So we need to make sure the UTXO is
+	// leased for basically forever.
+	return q.UpdateUTXOLease(ctx, UpdateUTXOLease{
+		LeaseOwner: finalLeaseOwner[:],
+		LeaseExpiry: sql.NullTime{
+			Time:  finalLeaseExpiry,
+			Valid: true,
+		},
+		Outpoint: anchorPointBytes,
+	})
 }
 
 // fetchAssetTransferInputs fetches all the inputs for a given transfer ID.

tapdb/assets_store_test.go

@@ -984,7 +984,9 @@ func TestAssetExportLog(t *testing.T) {
 			ProofSuffix: senderBlob,
 		}},
 	}
-	require.NoError(t, assetsStore.LogPendingParcel(ctx, spendDelta))
+	require.NoError(t, assetsStore.LogPendingParcel(
+		ctx, spendDelta, [32]byte{}, time.Now().Add(time.Hour),
+	))
 
 	assetID := inputAsset.ID()
 	proofs := map[asset.SerializedKey]*proof.AnnotatedProof{
@@ -1083,7 +1085,10 @@ func TestAssetExportLog(t *testing.T) {
 	// before.
 	require.Equal(t, numAssets+1, len(chainAssets))
 
-	var mutationFound bool
+	var (
+		mutationFound bool
+		inputLeased   bool
+	)
 	for _, chainAsset := range chainAssets {
 		// We should find the mutated asset with its _new_ script key
 		// and amount.
@@ -1101,8 +1106,22 @@ func TestAssetExportLog(t *testing.T) {
 			)
 			mutationFound = true
 		}
+
+		// The single UTXO we had at the beginning should now be leased
+		// for an hour.
+		if chainAsset.AnchorOutpoint == utxos[0].OutPoint {
+			require.Equal(
+				t, [32]byte{}, chainAsset.AnchorLeaseOwner,
+			)
+			require.True(
+				t,
+				chainAsset.AnchorLeaseExpiry.After(time.Now()),
+			)
+			inputLeased = true
+		}
 	}
 	require.True(t, mutationFound)
+	require.True(t, inputLeased)
 
 	// As a final check for the asset, we'll fetch its blob to ensure it's
 	// been updated on disk.

tapfreighter/chain_porter.go

@@ -872,7 +872,10 @@ func (p *ChainPorter) stateStep(currentPkg sendPackage) (*sendPackage, error) {
 
 		log.Infof("Committing pending parcel to disk")
 
-		err = p.cfg.ExportLog.LogPendingParcel(ctx, parcel)
+		err = p.cfg.ExportLog.LogPendingParcel(
+			ctx, parcel, defaultWalletLeaseIdentifier,
+			time.Now().Add(defaultBroadcastCoinLeaseDuration),
+		)
 		if err != nil {
 			return nil, fmt.Errorf("unable to write send pkg to "+
 				"disk: %v", err)

tapfreighter/interface.go

@@ -298,7 +298,8 @@ type ExportLog interface {
 	// LogPendingParcel marks an outbound parcel as pending on disk. This
 	// commits the set of changes to disk (the asset deltas) but doesn't
 	// mark the batched spend as being finalized.
-	LogPendingParcel(context.Context, *OutboundParcel) error
+	LogPendingParcel(context.Context, *OutboundParcel, [32]byte,
+		time.Time) error
 
 	// PendingParcels returns the set of parcels that haven't yet been
 	// finalized. This can be used to query the set of unconfirmed

tapfreighter/wallet.go

@@ -35,6 +35,13 @@ const (
 	// defaultCoinLeaseDuration is the default duration for which we lease
 	// managed UTXOs of asset outputs from the wallet.
 	defaultCoinLeaseDuration = 10 * time.Minute
+
+	// defaultBroadcastCoinLeaseDuration is the default duration for which
+	// we lease managed UTXOs of asset outputs from the wallet when we have
+	// broadcast a transaction that spends them. This represents a full year
+	// and avoids the same UTXO being used in another transaction if the
+	// confirmation of the first transaction takes a long time.
+	defaultBroadcastCoinLeaseDuration = 365 * 24 * time.Hour
 )
 
 var (