sweep: expand sweeper store to also save RBF-related info

This commit modifies the sweeper store to save a `TxRecord` in db
instead of an empty byte slice. This record will later be used to bring
RBF-awareness to our sweeper.

sweep/store.go

@@ -4,17 +4,19 @@ import (
 	"bytes"
 	"encoding/binary"
 	"errors"
+	"io"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/tlv"
 )
 
 var (
 	// txHashesBucketKey is the key that points to a bucket containing the
 	// hashes of all sweep txes that were published successfully.
 	//
-	// maps: txHash -> empty slice
+	// maps: txHash -> TxRecord
 	txHashesBucketKey = []byte("sweeper-tx-hashes")
 
 	// utxnChainPrefix is the bucket prefix for nursery buckets.
@@ -31,8 +33,90 @@ var (
 	byteOrder = binary.BigEndian
 
 	errNoTxHashesBucket = errors.New("tx hashes bucket does not exist")
+
+	// ErrTxNotFound is returned when querying using a txid that's not
+	// found in our db.
+	ErrTxNotFound = errors.New("tx not found")
 )
 
+// TxRecord specifies a record of a tx that's stored in the database.
+type TxRecord struct {
+	// Txid is the sweeping tx's txid, which is used as the key to store
+	// the following values.
+	Txid chainhash.Hash
+
+	// FeeRate is the fee rate of the sweeping tx, unit is sats/kw.
+	FeeRate uint64
+
+	// Fee is the fee of the sweeping tx, unit is sat.
+	Fee uint64
+
+	// Published indicates whether the tx has been published.
+	Published bool
+}
+
+// toTlvStream converts TxRecord into a tlv representation.
+func (t *TxRecord) toTlvStream() (*tlv.Stream, error) {
+	const (
+		// A set of tlv type definitions used to serialize TxRecord.
+		// We define it here instead of the head of the file to avoid
+		// naming conflicts.
+		//
+		// NOTE: A migration should be added whenever the existing type
+		// changes.
+		//
+		// NOTE: Txid is stored as the key, so it's not included here.
+		feeRateType tlv.Type = 0
+		feeType     tlv.Type = 1
+		boolType    tlv.Type = 2
+	)
+
+	return tlv.NewStream(
+		tlv.MakeBigSizeRecord(feeRateType, &t.FeeRate),
+		tlv.MakeBigSizeRecord(feeType, &t.Fee),
+		tlv.MakePrimitiveRecord(boolType, &t.Published),
+	)
+}
+
+// serializeTxRecord serializes a TxRecord based on tlv format.
+func serializeTxRecord(w io.Writer, tx *TxRecord) error {
+	// Create the tlv stream.
+	tlvStream, err := tx.toTlvStream()
+	if err != nil {
+		return err
+	}
+
+	// Encode the tlv stream.
+	var buf bytes.Buffer
+	if err := tlvStream.Encode(&buf); err != nil {
+		return err
+	}
+
+	// Write the tlv stream.
+	if _, err = w.Write(buf.Bytes()); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// deserializeTxRecord deserializes a TxRecord based on tlv format.
+func deserializeTxRecord(r io.Reader) (*TxRecord, error) {
+	var tx TxRecord
+
+	// Create the tlv stream.
+	tlvStream, err := tx.toTlvStream()
+	if err != nil {
+		return nil, err
+	}
+
+	if err := tlvStream.Decode(r); err != nil {
+		return nil, err
+	}
+
+	return &tx, nil
+}
+
 // SweeperStore stores published txes.
 type SweeperStore interface {
 	// IsOurTx determines whether a tx is published by us, based on its
@@ -40,7 +124,7 @@ type SweeperStore interface {
 	IsOurTx(hash chainhash.Hash) (bool, error)
 
 	// StoreTx stores a tx hash we are about to publish.
-	StoreTx(chainhash.Hash) error
+	StoreTx(*TxRecord) error
 
 	// ListSweeps lists all the sweeps we have successfully published.
 	ListSweeps() ([]chainhash.Hash, error)
@@ -83,6 +167,8 @@ func NewSweeperStore(db kvdb.Backend, chainHash *chainhash.Hash) (
 
 // migrateTxHashes migrates nursery finalized txes to the tx hashes bucket. This
 // is not implemented as a database migration, to keep the downgrade path open.
+//
+// TODO(yy): delete this function once nursery is removed.
 func migrateTxHashes(tx kvdb.RwTx, txHashesBucket kvdb.RwBucket,
 	chainHash *chainhash.Hash) error {
 
@@ -138,7 +224,24 @@ func migrateTxHashes(tx kvdb.RwTx, txHashesBucket kvdb.RwBucket,
 		log.Debugf("Inserting nursery tx %v in hash list "+
 			"(height=%v)", hash, byteOrder.Uint32(k))
 
-		return txHashesBucket.Put(hash[:], []byte{})
+		// Create the transaction record. Since this is an old record,
+		// we can assume it's already been published. Although it's
+		// possible to calculate the fees and fee rate used here, we
+		// skip it as it's unlikely we'd perform RBF on these old
+		// sweeping transactions.
+		tr := &TxRecord{
+			Txid:      hash,
+			Published: true,
+		}
+
+		// Serialize tx record.
+		var b bytes.Buffer
+		err = serializeTxRecord(&b, tr)
+		if err != nil {
+			return err
+		}
+
+		return txHashesBucket.Put(tr.Txid[:], b.Bytes())
 	})
 	if err != nil {
 		return err
@@ -148,15 +251,21 @@ func migrateTxHashes(tx kvdb.RwTx, txHashesBucket kvdb.RwBucket,
 }
 
 // StoreTx stores that we are about to publish a tx.
-func (s *sweeperStore) StoreTx(txid chainhash.Hash) error {
+func (s *sweeperStore) StoreTx(tr *TxRecord) error {
 	return kvdb.Update(s.db, func(tx kvdb.RwTx) error {
-
 		txHashesBucket := tx.ReadWriteBucket(txHashesBucketKey)
 		if txHashesBucket == nil {
 			return errNoTxHashesBucket
 		}
 
-		return txHashesBucket.Put(txid[:], []byte{})
+		// Serialize tx record.
+		var b bytes.Buffer
+		err := serializeTxRecord(&b, tr)
+		if err != nil {
+			return err
+		}
+
+		return txHashesBucket.Put(tr.Txid[:], b.Bytes())
 	}, func() {})
 }
 

sweep/store_mock.go

@@ -25,8 +25,8 @@ func (s *MockSweeperStore) IsOurTx(hash chainhash.Hash) (bool, error) {
 }
 
 // StoreTx stores a tx we are about to publish.
-func (s *MockSweeperStore) StoreTx(txid chainhash.Hash) error {
+func (s *MockSweeperStore) StoreTx(tr *TxRecord) error {
-	s.ourTxes[txid] = struct{}{}
+	s.ourTxes[tr.Txid] = struct{}{}
 
 	return nil
 }

sweep/store_test.go

@@ -1,6 +1,7 @@
 package sweep
 
 import (
+	"bytes"
 	"testing"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
@@ -50,11 +51,13 @@ func testStore(t *testing.T, createStore func() (SweeperStore, error)) {
 		},
 	})
 
-	err = store.StoreTx(tx1.TxHash())
+	tr1 := &TxRecord{
-	if err != nil {
+		Txid: tx1.TxHash(),
-		t.Fatal(err)
 	}
 
+	err = store.StoreTx(tr1)
+	require.NoError(t, err)
+
 	// Notify publication of tx2
 	tx2 := wire.MsgTx{}
 	tx2.AddTxIn(&wire.TxIn{
@@ -63,11 +66,13 @@ func testStore(t *testing.T, createStore func() (SweeperStore, error)) {
 		},
 	})
 
-	err = store.StoreTx(tx2.TxHash())
+	tr2 := &TxRecord{
-	if err != nil {
+		Txid: tx2.TxHash(),
-		t.Fatal(err)
 	}
 
+	err = store.StoreTx(tr2)
+	require.NoError(t, err)
+
 	// Recreate the sweeper store
 	store, err = createStore()
 	if err != nil {
@@ -76,30 +81,18 @@ func testStore(t *testing.T, createStore func() (SweeperStore, error)) {
 
 	// Assert that both txes are recognized as our own.
 	ours, err := store.IsOurTx(tx1.TxHash())
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+	require.True(t, ours, "expected tx to be ours")
-	}
-	if !ours {
-		t.Fatal("expected tx to be ours")
-	}
 
 	ours, err = store.IsOurTx(tx2.TxHash())
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+	require.True(t, ours, "expected tx to be ours")
-	}
-	if !ours {
-		t.Fatal("expected tx to be ours")
-	}
 
 	// An different hash should be reported as not being ours.
 	var unknownHash chainhash.Hash
 	ours, err = store.IsOurTx(unknownHash)
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+	require.False(t, ours, "expected tx to not be ours")
-	}
-	if ours {
-		t.Fatal("expected tx to be not ours")
-	}
 
 	txns, err := store.ListSweeps()
 	require.NoError(t, err, "unexpected error")
@@ -110,16 +103,38 @@ func testStore(t *testing.T, createStore func() (SweeperStore, error)) {
 		tx1.TxHash(): true,
 		tx2.TxHash(): true,
 	}
-
+	require.Len(t, txns, len(expected))
-	if len(txns) != len(expected) {
-		t.Fatalf("expected: %v sweeps, got: %v", len(expected),
-			len(txns))
-	}
 
 	for _, tx := range txns {
 		_, ok := expected[tx]
-		if !ok {
+		require.Truef(t, ok, "unexpected txid returned: %v", tx)
-			t.Fatalf("unexpected tx: %v", tx)
-		}
 	}
 }
+
+// TestTxRecord asserts that the serializeTxRecord and deserializeTxRecord
+// behave as expected.
+func TestTxRecord(t *testing.T) {
+	t.Parallel()
+
+	// Create a testing record.
+	//
+	// NOTE: Txid is omitted because it is not serialized.
+	tr := &TxRecord{
+		FeeRate:   1000,
+		Fee:       10000,
+		Published: true,
+	}
+
+	var b bytes.Buffer
+
+	// Assert we can serialize the record.
+	err := serializeTxRecord(&b, tr)
+	require.NoError(t, err)
+
+	// Assert we can deserialize the record.
+	result, err := deserializeTxRecord(&b)
+	require.NoError(t, err)
+
+	// Assert the deserialized record is equal to the original.
+	require.Equal(t, tr, result)
+}

sweep/sweeper.go

@@ -1163,7 +1163,7 @@ func (s *UtxoSweeper) sweep(inputs inputSet, feeRate chainfee.SatPerKWeight,
 	}
 
 	// Create sweep tx.
-	tx, _, err := createSweepTx(
+	tx, fee, err := createSweepTx(
 		inputs, nil, s.currentOutputScript, uint32(currentHeight),
 		feeRate, s.cfg.MaxFeeRate.FeePerKWeight(), s.cfg.Signer,
 	)
@@ -1171,12 +1171,18 @@ func (s *UtxoSweeper) sweep(inputs inputSet, feeRate chainfee.SatPerKWeight,
 		return fmt.Errorf("create sweep tx: %w", err)
 	}
 
+	tr := &TxRecord{
+		Txid:    tx.TxHash(),
+		FeeRate: uint64(feeRate),
+		Fee:     uint64(fee),
+	}
+
 	// Add tx before publication, so that we will always know that a spend
 	// by this tx is ours. Otherwise if the publish doesn't return, but did
 	// publish, we loose track of this tx. Even republication on startup
 	// doesn't prevent this, because that call returns a double spend error
 	// then and would also not add the hash to the store.
-	err = s.cfg.Store.StoreTx(tx.TxHash())
+	err = s.cfg.Store.StoreTx(tr)
 	if err != nil {
 		return fmt.Errorf("store tx: %w", err)
 	}
@@ -1197,6 +1203,16 @@ func (s *UtxoSweeper) sweep(inputs inputSet, feeRate chainfee.SatPerKWeight,
 		return err
 	}
 
+	// Mark this tx in db once successfully published.
+	//
+	// NOTE: this will behave as an overwrite, which is fine as the record
+	// is small.
+	tr.Published = true
+	err = s.cfg.Store.StoreTx(tr)
+	if err != nil {
+		return fmt.Errorf("store tx: %w", err)
+	}
+
 	// If there's no error, remove the output script. Otherwise keep it so
 	// that it can be reused for the next transaction and causes no address
 	// inflation.