Skip to content

Add Multi-RFQ Send #1613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions fn/send.go
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package fn

import "context"

// SendOrQuit attempts to and a message through channel c. If this succeeds,
// then bool is returned. Otherwise if a quit signal is received first, then
// false is returned.
Expand All @@ -12,6 +14,19 @@ func SendOrQuit[T any, Q any](c chan<- T, msg T, quit chan Q) bool {
}
}

// SendOrDone attempts to and a message through channel c. If this succeeds,
// then bool is returned. Otherwise if a ctx done signal is received first, then
// false is returned.
func SendOrDone[T any](ctx context.Context, c chan<- T, msg T) bool {
select {
case c <- msg:
return true

case <-ctx.Done():
return false
}
}

// SendAll attempts to send all messages through channel c.
//
// TODO(roasbeef): add non-blocking variant?
Expand Down
3 changes: 0 additions & 3 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,3 @@ replace google.golang.org/protobuf => github.com/lightninglabs/protobuf-go-hex-d
// We are using a fork of the migration library with custom functionality that
// did not yet make it into the upstream repository.
replace github.com/golang-migrate/migrate/v4 => github.com/lightninglabs/migrate/v4 v4.18.2-9023d66a-fork-pr-2

// Note this is a temproary replace and will be removed when taprpc is tagged.
replace github.com/lightninglabs/taproot-assets/taprpc => ./taprpc
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will make future PRs that change anything in the taprpc package super hard (needs to be two PRs). So I think we decided to leave this in. But we should update the version on line 32 regularly so external consumers will pull in the correct version.

2 changes: 2 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -1147,6 +1147,8 @@ github.com/lightninglabs/neutrino/cache v1.1.2 h1:C9DY/DAPaPxbFC+xNNEI/z1SJY9GS3
github.com/lightninglabs/neutrino/cache v1.1.2/go.mod h1:XJNcgdOw1LQnanGjw8Vj44CvguYA25IMKjWFZczwZuo=
github.com/lightninglabs/protobuf-go-hex-display v1.34.2-hex-display h1:w7FM5LH9Z6CpKxl13mS48idsu6F+cEZf0lkyiV+Dq9g=
github.com/lightninglabs/protobuf-go-hex-display v1.34.2-hex-display/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
github.com/lightninglabs/taproot-assets/taprpc v1.0.9 h1:Rr/B/6QURBwflA6LNFWuQCBOP5Xh3WbHQf/dNf8tA98=
github.com/lightninglabs/taproot-assets/taprpc v1.0.9/go.mod h1:c8gTEcKEUoUPVChgZNwqTL1hss7UWa5FDeObr8WBzQk=
github.com/lightningnetwork/lightning-onion v1.2.1-0.20240712235311-98bd56499dfb h1:yfM05S8DXKhuCBp5qSMZdtSwvJ+GFzl94KbXMNB1JDY=
github.com/lightningnetwork/lightning-onion v1.2.1-0.20240712235311-98bd56499dfb/go.mod h1:c0kvRShutpj3l6B9WtTsNTBUtjSmjZXbJd9ZBRQOSKI=
github.com/lightningnetwork/lnd v0.19.2-beta h1:3SKVrKYFY4IJLlrMf7cDzZcBeT+MxjI9Xy6YpY+EEX4=
Expand Down
2 changes: 1 addition & 1 deletion itest/rfq_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -382,7 +382,7 @@ func testRfqAssetSellHtlcIntercept(t *harnessTest) {
}

htlcCustomRecords := rfqmsg.NewHtlc(
assetAmounts, fn.Some(acceptedQuoteId),
assetAmounts, fn.Some(acceptedQuoteId), fn.None[[]rfqmsg.ID](),
)

// Convert the custom records to a TLV map for inclusion in
Expand Down
4 changes: 3 additions & 1 deletion rfq/order.go
Original file line number Diff line number Diff line change
Expand Up @@ -279,7 +279,9 @@ func (c *AssetSalePolicy) GenerateInterceptorResponse(
// Include the asset balance in the HTLC record.
htlcBalance := rfqmsg.NewAssetBalance(assetID, amt)
htlcRecord := rfqmsg.NewHtlc(
[]*rfqmsg.AssetBalance{htlcBalance}, fn.Some(c.AcceptedQuoteId),
[]*rfqmsg.AssetBalance{htlcBalance},
fn.Some(c.AcceptedQuoteId),
fn.None[[]rfqmsg.ID](),
)

customRecords, err := lnwire.ParseCustomRecords(htlcRecord.Bytes())
Expand Down
5 changes: 3 additions & 2 deletions rfqmsg/custom_channel_data.go
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@ type JsonAssetTranche struct {
// JsonHtlc is a struct that represents the asset information that can be
// transferred via an HTLC.
type JsonHtlc struct {
Balances []*JsonAssetTranche `json:"balances"`
RfqID string `json:"rfq_id"`
Balances []*JsonAssetTranche `json:"balances"`
RfqID string `json:"rfq_id"`
AvailableRfqIDs []string `json:"available_rfq_ids,omitempty"`
}
143 changes: 140 additions & 3 deletions rfqmsg/records.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,13 @@ const (
// be from the same asset group but from different tranches to be
// encoded as an individual record.
MaxNumOutputs = 2048

// MaxSendPaymentQuotes is the upper limit of quotes that may be
// acquired by the SendPayment RPC endpoint in order to carry out the
// payment. Acquiring more quotes than whatever is defined below is
// possible, but may cause performance and network I/O bottlenecks as
// a lot of RFQ messages are getting involved.
MaxSendPaymentQuotes = 50
)

var (
Expand All @@ -42,13 +49,107 @@ type (
// encode an RFQ id within the custom records of an HTLC record on the
// wire.
HtlcRfqIDType = tlv.TlvType65538

// AvailableRfqIDsType is the type alias for the TLV type that is used
// to encode the list of available RFQ IDs that can be used for an HTLC.
// This list is only meant to be handled by the complementary LND
// instance via the AuxTrafficShaper hooks.
AvailableRfqIDsType = tlv.TlvType65540
)

// SomeRfqIDRecord creates an optional record that represents an RFQ ID.
func SomeRfqIDRecord(id ID) tlv.OptionalRecordT[HtlcRfqIDType, ID] {
return tlv.SomeRecordT(tlv.NewPrimitiveRecord[HtlcRfqIDType, ID](id))
}

// HtlcRfqIDs is a helper wrapper around the array of IDs that can be encoded as
// records.
type HtlcRfqIDs struct {
// IDs is a list of RFQ IDs that are associated with the HTLC.
IDs []ID
}

// Record returns the tlv record of RfqIDs.
func (r *HtlcRfqIDs) Record() tlv.Record {
size := func() uint64 {
var (
buf bytes.Buffer
scratch [8]byte
)
err := encodeAvailableIDs(&buf, r, &scratch)
if err != nil {
panic(fmt.Sprintf("unable to encode RFQ IDs: %v", err))
}
return uint64(buf.Len())
}

// Note that we set the type here as zero, as when used with a
// tlv.RecordT, the type param will be used as the type.
return tlv.MakeDynamicRecord(
0, r, size, encodeAvailableIDs, decodeRfqIDs,
)
}

// encodeAvailableIDs encodes the RfqIDs struct on the wire.
func encodeAvailableIDs(w io.Writer, val any, buf *[8]byte) error {
if rfqIDs, ok := val.(*HtlcRfqIDs); ok {
ids := rfqIDs.IDs
num := uint64(len(ids))
if err := tlv.WriteVarInt(w, num, buf); err != nil {
return err
}
for _, id := range ids {
idCopy := id
if err := IdEncoder(w, &idCopy, buf); err != nil {
return err
}
}

return nil
}
return tlv.NewTypeForEncodingErr(val, "*RfqIDs")
}

// decodeRfqIDs decodes the RfqIDs from the wire.
func decodeRfqIDs(r io.Reader, val any, buf *[8]byte, _ uint64) error {
if ids, ok := val.(*HtlcRfqIDs); ok {
num, err := tlv.ReadVarInt(r, buf)
if err != nil {
return err
}

if num > MaxSendPaymentQuotes {
return fmt.Errorf("available RFQ IDs array length too "+
"big, max allowed is %v, got %v",
MaxSendPaymentQuotes, num)
}

list := make([]ID, num)
for i := uint64(0); i < num; i++ {
var id ID
if err := IdDecoder(r, &id, buf, 32); err != nil {
return err
}
list[i] = id
}

ids.IDs = list

return nil
}
return tlv.NewTypeForDecodingErr(val, "*RfqIDs", 0, 0)
}

// SomeRfqIDsRecord creates an optional record that represents the array of
// available RFQ IDs.
func SomeRfqIDsRecord(
ids []ID) tlv.OptionalRecordT[AvailableRfqIDsType, HtlcRfqIDs] {

return tlv.SomeRecordT(tlv.NewRecordT[AvailableRfqIDsType, HtlcRfqIDs](
HtlcRfqIDs{IDs: ids},
))
}

// Htlc is a record that represents the capacity change related to an in-flight
// HTLC. This entails all the (asset_id, amount) tuples and other information
// that we may need to be able to update the TAP portion of a commitment
Expand All @@ -59,21 +160,33 @@ type Htlc struct {

// RfqID is the RFQ ID that corresponds to the HTLC.
RfqID tlv.OptionalRecordT[HtlcRfqIDType, ID]

// AvailableRfqIDs is a list of RFQ IDs that may be used for this HTLC.
// This is used by the traffic shaper when querying for available
// bandwidth.
AvailableRfqIDs tlv.OptionalRecordT[AvailableRfqIDsType, HtlcRfqIDs]
}

// NewHtlc creates a new Htlc record with the given funded assets.
func NewHtlc(amounts []*AssetBalance, rfqID fn.Option[ID]) *Htlc {
func NewHtlc(amounts []*AssetBalance, rfqID fn.Option[ID],
availableRfqIDs fn.Option[[]ID]) *Htlc {

htlc := &Htlc{
Amounts: tlv.NewRecordT[HtlcAmountRecordType](
AssetBalanceListRecord{
Balances: amounts,
},
),
}

rfqID.WhenSome(func(id ID) {
htlc.RfqID = SomeRfqIDRecord(id)
})

availableRfqIDs.WhenSome(func(ids []ID) {
htlc.AvailableRfqIDs = SomeRfqIDsRecord(ids)
})

return htlc
}

Expand Down Expand Up @@ -135,6 +248,12 @@ func (h *Htlc) Records() []tlv.Record {
records = append(records, r.Record())
})

h.AvailableRfqIDs.WhenSome(
func(r tlv.RecordT[AvailableRfqIDsType, HtlcRfqIDs]) {
records = append(records, r.Record())
},
)

return records
}

Expand All @@ -154,11 +273,13 @@ func (h *Htlc) Encode(w io.Writer) error {
// Decode deserializes the Htlc from the given io.Reader.
func (h *Htlc) Decode(r io.Reader) error {
rfqID := h.RfqID.Zero()
rfqIDs := h.AvailableRfqIDs.Zero()

// Create the tlv stream.
tlvStream, err := tlv.NewStream(
h.Amounts.Record(),
rfqID.Record(),
rfqIDs.Record(),
)
if err != nil {
return err
Expand All @@ -173,6 +294,10 @@ func (h *Htlc) Decode(r io.Reader) error {
h.RfqID = tlv.SomeRecordT(rfqID)
}

if val, ok := typeMap[h.AvailableRfqIDs.TlvType()]; ok && val == nil {
h.AvailableRfqIDs = tlv.SomeRecordT(rfqIDs)
}

return nil
}

Expand All @@ -198,6 +323,13 @@ func (h *Htlc) AsJson() ([]byte, error) {
j.RfqID = hex.EncodeToString(id[:])
})

h.AvailableRfqIDs.ValOpt().WhenSome(func(rfqIDs HtlcRfqIDs) {
j.AvailableRfqIDs = make([]string, len(rfqIDs.IDs))
for idx, id := range rfqIDs.IDs {
j.AvailableRfqIDs[idx] = hex.EncodeToString(id[:])
}
})

for idx, balance := range h.Balances() {
j.Balances[idx] = &JsonAssetTranche{
AssetID: hex.EncodeToString(balance.AssetID.Val[:]),
Expand Down Expand Up @@ -235,8 +367,9 @@ func HtlcFromCustomRecords(records lnwire.CustomRecords) (*Htlc, error) {
// the custom records that we'd expect an asset HTLC to carry.
func HasAssetHTLCCustomRecords(records lnwire.CustomRecords) bool {
var (
amountType HtlcAmountRecordType
rfqIDType HtlcRfqIDType
amountType HtlcAmountRecordType
rfqIDType HtlcRfqIDType
availableIDs AvailableRfqIDsType
)
for key := range records {
if key == uint64(amountType.TypeVal()) {
Expand All @@ -246,6 +379,10 @@ func HasAssetHTLCCustomRecords(records lnwire.CustomRecords) bool {
if key == uint64(rfqIDType.TypeVal()) {
return true
}

if key == uint64(availableIDs.TypeVal()) {
return true
}
}

return false
Expand Down
39 changes: 35 additions & 4 deletions rfqmsg/records_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,11 @@ type htlcTestCase struct {

type DummyChecker struct{}

var (
dummyRFQ1 = ID(bytes.Repeat([]byte{1}, 32))
dummyRFQ2 = ID(bytes.Repeat([]byte{2}, 32))
)

func MockAssetMatchesSpecifier(_ context.Context, specifier asset.Specifier,
id asset.ID) (bool, error) {

Expand Down Expand Up @@ -112,7 +117,7 @@ func TestHtlc(t *testing.T) {
name: "HTLC with balance asset",
htlc: NewHtlc([]*AssetBalance{
NewAssetBalance([32]byte{1}, 1000),
}, fn.None[ID]()),
}, fn.None[ID](), fn.None[[]ID]()),
expectRecords: true,
//nolint:lll
expectedJSON: `{
Expand All @@ -131,7 +136,7 @@ func TestHtlc(t *testing.T) {
NewAssetBalance([32]byte{1}, 1000),
NewAssetBalance([32]byte{1}, 2000),
NewAssetBalance([32]byte{2}, 5000),
}, fn.None[ID]()),
}, fn.None[ID](), fn.None[[]ID]()),
sumBalances: map[asset.ID]rfqmath.BigInt{
[32]byte{1}: rfqmath.NewBigIntFromUint64(3000),
[32]byte{2}: rfqmath.NewBigIntFromUint64(5000),
Expand All @@ -143,7 +148,29 @@ func TestHtlc(t *testing.T) {
htlc: NewHtlc([]*AssetBalance{
NewAssetBalance([32]byte{1}, 1000),
NewAssetBalance([32]byte{2}, 2000),
}, fn.Some(ID{0, 1, 2, 3, 4, 5, 6, 7})),
}, fn.Some(dummyRFQ1), fn.None[[]ID]()),
expectRecords: true,
//nolint:lll
expectedJSON: `{
"balances": [
{
"asset_id": "0100000000000000000000000000000000000000000000000000000000000000",
"amount": 1000
},
{
"asset_id": "0200000000000000000000000000000000000000000000000000000000000000",
"amount": 2000
}
],
"rfq_id": "0101010101010101010101010101010101010101010101010101010101010101"
}`,
},
{
name: "channel with multiple balance assets",
htlc: NewHtlc([]*AssetBalance{
NewAssetBalance([32]byte{1}, 1000),
NewAssetBalance([32]byte{2}, 2000),
}, fn.None[ID](), fn.Some([]ID{dummyRFQ1, dummyRFQ2})),
expectRecords: true,
//nolint:lll
expectedJSON: `{
Expand All @@ -157,7 +184,11 @@ func TestHtlc(t *testing.T) {
"amount": 2000
}
],
"rfq_id": "0001020304050607000000000000000000000000000000000000000000000000"
"rfq_id": "",
"available_rfq_ids": [
"0101010101010101010101010101010101010101010101010101010101010101",
"0202020202020202020202020202020202020202020202020202020202020202"
]
}`,
},
}
Expand Down
Loading
Loading