diff --git a/sweep/store.go b/sweep/store.go index 72d7853a4..375436ca3 100644 --- a/sweep/store.go +++ b/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() {}) } diff --git a/sweep/store_mock.go b/sweep/store_mock.go index c8b9652f4..8263cb3a8 100644 --- a/sweep/store_mock.go +++ b/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 { - s.ourTxes[txid] = struct{}{} +func (s *MockSweeperStore) StoreTx(tr *TxRecord) error { + s.ourTxes[tr.Txid] = struct{}{} return nil } diff --git a/sweep/store_test.go b/sweep/store_test.go index c07583dcc..4abf65871 100644 --- a/sweep/store_test.go +++ b/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()) - if err != nil { - t.Fatal(err) + tr1 := &TxRecord{ + Txid: tx1.TxHash(), } + 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()) - if err != nil { - t.Fatal(err) + tr2 := &TxRecord{ + Txid: tx2.TxHash(), } + 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 { - t.Fatal(err) - } - if !ours { - t.Fatal("expected tx to be ours") - } + require.NoError(t, err) + require.True(t, ours, "expected tx to be ours") ours, err = store.IsOurTx(tx2.TxHash()) - if err != nil { - t.Fatal(err) - } - if !ours { - t.Fatal("expected tx to be ours") - } + require.NoError(t, err) + require.True(t, ours, "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 { - t.Fatal(err) - } - if ours { - t.Fatal("expected tx to be not ours") - } + require.NoError(t, err) + require.False(t, ours, "expected tx to not be 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, } - - if len(txns) != len(expected) { - t.Fatalf("expected: %v sweeps, got: %v", len(expected), - len(txns)) - } + require.Len(t, txns, len(expected)) for _, tx := range txns { _, ok := expected[tx] - if !ok { - t.Fatalf("unexpected tx: %v", tx) - } + require.Truef(t, ok, "unexpected txid returned: %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) +} diff --git a/sweep/sweeper.go b/sweep/sweeper.go index c1981eed2..f3920a414 100644 --- a/sweep/sweeper.go +++ b/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.