From 3dee84950254939a7477018b9257e343585dbe9d Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Tue, 18 May 2021 19:38:48 -0700 Subject: [PATCH 01/12] clientdb: permit sub-buckets in main sidecar bucket A prior check is now invalidated as the sidecar bucket houses a sub-bucket that we use to store the set of bid information related to a sidecar channel. --- clientdb/sidecar.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/clientdb/sidecar.go b/clientdb/sidecar.go index ac30fad..23cae76 100644 --- a/clientdb/sidecar.go +++ b/clientdb/sidecar.go @@ -206,9 +206,11 @@ func (db *DB) Sidecars() ([]*sidecar.Ticket, error) { } return sidecarBucket.ForEach(func(k, v []byte) error { - // We don't expect any sub-buckets with sidecars. + // The main sidecar bucket has a sub-bucket that's used + // to store order bid information, so we'll skip this + // bucket when attempting to read out all the tickets. if v == nil { - return fmt.Errorf("nil value for key %x", k) + return nil } s, err := readSidecar(sidecarBucket, k) From d60b991b6f8ba81e4d9003c06d72bf3b2fc2d4de Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Tue, 18 May 2021 19:45:04 -0700 Subject: [PATCH 02/12] sidecar: insert missing continue during auto sidecar restart A continue was missing that would cause the loop to panic below as in certain cases the recipient isn't yet present (provider just restarted). --- sidecar_acceptor.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sidecar_acceptor.go b/sidecar_acceptor.go index c1ef9a6..2ce6cf1 100644 --- a/sidecar_acceptor.go +++ b/sidecar_acceptor.go @@ -174,6 +174,8 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { "for sidecar: %w", err) } + continue + // If the ticket has no recipient or isn't in the expecting // state, then we can safely skip it. case ticket.State != sidecar.StateExpectingChannel: From 3ca1be322ecbcea9c8d2bd4cacd8f7076a9e3c26 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 19 May 2021 21:00:59 -0700 Subject: [PATCH 03/12] sidecar: allow recipient ID to be derived using base offer ticket In this commit, we fix a bug in the retransmission case for the provider. Before this commit, the provider would attempt to re-send the initial offer ticket on restart if it hadn't yet received the registered ticket. This fails as the recipient's stream ID is derived from their pubkey information that's only contained in the registered ticket. To fix this, we'll simply use the sha512 hash of the offer sig since it's known when the ticket is initially created. --- auto_sidecar.go | 45 ++++++++++++++++++++++++++------------------- 1 file changed, 26 insertions(+), 19 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index c79652c..c15c666 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -3,6 +3,7 @@ package pool import ( "bytes" "context" + "crypto/sha512" "errors" "fmt" @@ -53,21 +54,24 @@ func deriveProviderStreamID(ticket *sidecar.Ticket) ([64]byte, error) { // deriveRecipientStreamID derives the stream ID of the cipher box that the // provider of the sidecar ticket will use to send messages to the receiver. -func deriveRecipientStreamID(ticket *sidecar.Ticket) [64]byte { - receiverMultisig := ticket.Recipient.NodePubKey.SerializeCompressed() - receiverNode := ticket.Recipient.MultiSigPubKey.SerializeCompressed() - - // The stream ID will be the concentration of the receiver's multi-sig - // and node keys, ignoring the first byte of each key that essentially - // communicates parity information. - var ( - streamID [64]byte - n int +func deriveRecipientStreamID(ticket *sidecar.Ticket) ([64]byte, error) { + // In order to ensure our retransmission case for the provider works + // (on start up, it resends the offered ticket in case it got the + // registered but didn't commit to disk), the provider needs to be able + // to compute the recipient's stream ID using the base offered ticket. + // + // To enable this, we'll use the sha256 of the offer sig as this is + // static for the lifetime of the entire ticket. + wireSig, err := lnwire.NewSigFromRawSignature( + ticket.Offer.SigOfferDigest.Serialize(), ) - n += copy(streamID[:], receiverMultisig[1:]) - copy(streamID[n:], receiverNode[1:]) + if err != nil { + return [64]byte{}, err + } - return streamID + streamID := sha512.Sum512(wireSig[:]) + + return streamID, nil } // deriveStreamID derives corresponding stream ID for the provider of the @@ -77,7 +81,7 @@ func deriveStreamID(ticket *sidecar.Ticket, provider bool) ([64]byte, error) { return deriveProviderStreamID(ticket) } - return deriveRecipientStreamID(ticket), nil + return deriveRecipientStreamID(ticket) } // sendSidecarPkt attempts to send a sidecar packet to the opposite party using @@ -167,15 +171,19 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { // Before we enter our main read loop below, we'll attempt to re-create // out mailbox as the recipient. - recipientStreamID := deriveRecipientStreamID( + recipientStreamID, err := deriveRecipientStreamID( localTicket, ) + if err != nil { + log.Errorf("unable to derive recipient ID: %v", err) + return + } log.Infof("Creating receiver reply mailbox for ticket=%x, "+ "stream_id=%x", startingPkt.ReceiverTicket.ID[:], recipientStreamID[:]) - err := a.client.InitTicketCipherBox( + err = a.client.InitTicketCipherBox( context.Background(), recipientStreamID, startingPkt.ReceiverTicket, ) @@ -200,6 +208,7 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { startingPkt.ReceiverTicket, false, ) if err != nil { + // TODO(roasbeef): back off then retry? log.Error(err) return } @@ -337,7 +346,7 @@ func (a *SidecarAcceptor) stateStepRecipient(pkt *SidecarPacket, // it. default: return nil, fmt.Errorf("unhandled receiver state transition "+ - "for ticket=%v, state=%v", pkt.ProviderTicket.ID[:], + "for ticket=%x, state=%v", pkt.ProviderTicket.ID[:], pkt.ProviderTicket.State) } } @@ -420,8 +429,6 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, for { priorState := currentState - log.Infof("step=%v", currentState) - newPktState, err := a.stateStepProvider(&SidecarPacket{ CurrentState: currentState, ReceiverTicket: newTicket, From 2e9994138923e2dbb6ff40d092ca23a7ae96c31e Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 20 May 2021 21:27:51 -0700 Subject: [PATCH 04/12] sidecar: abstract automated negotiation into new struct In this commit, we extract the existing automated sidecar negotiation functionality into a new struct. This is strictly a refactoring change intended to allow the core code to be more easily unit tested as all interaction now behind a set of interfaces permitting greater testability via mocks. --- auto_sidecar.go | 237 ++++++++++++++++++++++++++------------------ sidecar_acceptor.go | 200 +++++++++++++++++++++++++++++++------ 2 files changed, 312 insertions(+), 125 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index c15c666..a6ed2e7 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -1,21 +1,45 @@ package pool import ( - "bytes" "context" "crypto/sha512" "errors" "fmt" + "sync" "github.com/lightninglabs/pool/account" "github.com/lightninglabs/pool/clientdb" "github.com/lightninglabs/pool/order" "github.com/lightninglabs/pool/sidecar" + "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lnwire" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) +// MailBox is an interface that abstracts over the HashMail functionality to +// represent a generic mailbox that both sides will use to communicate with +// each other. +type MailBox interface { + // RecvSidecarPkt attempts to receive a new sidecar packet from the + // relevant mailbox defined by the ticket and sidecar ticket role. + RecvSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, + provider bool) (*sidecar.Ticket, error) + + // SendSidecarPkt attempts to send the specified sidecar ticket to the + // party designated by the provider bool. + SendSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, + provider bool) error + + // InitSidecarMailbox attempts to create the mailbox with the given + // stream ID using the sidecar ticket authentication mechanism. If the + // mailbox already exists, then a nil error is to be returned. + InitSidecarMailbox(streamID [64]byte, ticket *sidecar.Ticket) error + + // InitAcctMailbox attempts to create the mailbox with the given stream + // ID using account signature authentication mechanism. If the mailbox + // already exists, then a nil error is to be returned. + InitAcctMailbox(streamID [64]byte, pubKey *keychain.KeyDescriptor) error +} + // SidecarPacket encapsulates the current state of an auto sidecar negotiator. // Note that the state of the negotiator, and the ticket may differ, this is // what will trigger a state transition. @@ -35,7 +59,6 @@ type SidecarPacket struct { // deriveProviderStreamID derives the stream ID of the provider's cipher box, // we'll use this to allow the recipient to send messages to the provider. func deriveProviderStreamID(ticket *sidecar.Ticket) ([64]byte, error) { - var streamID [64]byte // This stream ID will simply be the fixed 64-byte signature of our @@ -84,79 +107,109 @@ func deriveStreamID(ticket *sidecar.Ticket, provider bool) ([64]byte, error) { return deriveRecipientStreamID(ticket) } -// sendSidecarPkt attempts to send a sidecar packet to the opposite party using -// their registered cipherbox stream. -func (a *SidecarAcceptor) sendSidecarPkt(pkt *sidecar.Ticket, - provider bool) error { +// SidecarDriver houses a series of methods needed to drive a given sidecar +// channel towards completion. +type SidecarDriver interface { + // ValidateOrderedTicketctx attempts to validate that a given ticket + // has rpoerly transitioned to the ordered state. + ValidateOrderedTicket(tkt *sidecar.Ticket) error - var ticketBuf bytes.Buffer - err := sidecar.SerializeTicket(&ticketBuf, pkt) - if err != nil { - return err - } + // ExpectChannel is called by the receiver of a channel once the + // negotiation process has been finalized, and they need to await a new + // channel funding flow initiated by the auctioneer server. + ExpectChannel(ctx context.Context, tkt *sidecar.Ticket) error - streamID, err := deriveStreamID(pkt, provider) - if err != nil { - return err - } + // UpdateSidecar writes the passed sidecar ticket to persistent + // storage. + UpdateSidecar(tkt *sidecar.Ticket) error - target := "receiver" - if provider { - target = "provider" - } - - log.Infof("Sending ticket(state=%v, id=%x) to %v stream_id=%x", - pkt.State, pkt.ID[:], target, streamID[:]) - - return a.client.SendCipherBoxMsg( - context.Background(), streamID, ticketBuf.Bytes(), - ) + // SubmitOrder submits a bid derived from the sidecar ticket, account, + // and bid template to the auctioneer. + SubmitSidecarOrder(*sidecar.Ticket, *order.Bid, + *account.Account) (*sidecar.Ticket, error) } -// recvSidecarPkt attempts to receive a new sidecar packet from the opposite -// party using their registered cipherbox stream. -func (a *SidecarAcceptor) recvSidecarPkt(ticket *sidecar.Ticket, - provider bool) (*sidecar.Ticket, error) { +// AutoAcceptorConfig houses all the functionality the sidecar negotiator needs +// to carry out its duties. +type AutoAcceptorConfig struct { + // Provider denotes if the negotiator is the provider or not. + Provider bool - streamID, err := deriveStreamID(ticket, provider) - if err != nil { - return nil, err + // ProviderBid is the provider's bid template. + ProviderBid *order.Bid + + // ProviderAccount points to the active account of the provider of the + // ticket. + ProviderAccount *account.Account + + // StartingPkt is the starting packet, or the starting state from the + // PoV of the negotiator. + StartingPkt *SidecarPacket + + // Drive contains functionality needed to drive a new sidecar ticket + // towards completion. + Driver SidecarDriver + + // MailBox is used to allow negotiators to send messages back and forth + // to each other. + MailBox MailBox + + // Quit is a global quit channel that all spawned goroutines will + // select on select on. + Quit chan struct{} +} + +// SidecarNegotiator is a sub-system that uses a mailbox abstraction between a +// provider and recipient of a sidecar channel to complete the manual steps in +// automated manner. +type SidecarNegotiator struct { + cfg AutoAcceptorConfig + + wg sync.WaitGroup +} + +// NewSidecarNegotiator returns a new instance of the sidecar negotiator given +// a valid config. +func NewSidecarNegotiator(cfg AutoAcceptorConfig) *SidecarNegotiator { + return &SidecarNegotiator{ + cfg: cfg, } +} - log.Infof("Waiting for ticket (id=%x) using stream_id=%x, provider=%v", - ticket.ID[:], streamID[:], provider) +// Start kicks off the set of goroutines needed for the sidecar channel to be +// negotiated. +func (a *SidecarNegotiator) Start() error { + // In order to ensure we can exit properly if signalled, we'll launch a + // goroutine that will cancel a global context if we need to exit. + a.wg.Add(1) ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + go func() { + defer a.wg.Done() - msg, err := a.client.RecvCipherBoxMsg(ctx, streamID) - if err != nil { - return nil, fmt.Errorf("unable to recv cipher box "+ - "msg: %w", err) + <-a.cfg.Quit + + cancel() + }() + + if a.cfg.Provider { + a.wg.Add(1) + go a.autoSidecarProvider( + ctx, a.cfg.StartingPkt, a.cfg.ProviderBid, a.cfg.ProviderAccount, + ) + } else { + a.wg.Add(1) + go a.autoSidecarReceiver(ctx, a.cfg.StartingPkt) } - log.Infof("Receive new message for ticket (id=%x) "+ - "via stream_id=%x, provider=%v", ticket.ID[:], streamID, - provider) - - return sidecar.DeserializeTicket(bytes.NewReader(msg)) -} - -// isErrAlreadyExists returns true if the passed error is the "already exists" -// error within the error wrapped error which is returned by the hash mail -// server when a stream we're attempting to create already exists. -func isErrAlreadyExists(err error) bool { - statusCode, ok := status.FromError(err) - if !ok { - return false - } - - return statusCode.Code() == codes.AlreadyExists + return nil } // autoSidecarReceiver is a goroutine that will attempt to advance a new // sidecar ticket through the process until it reaches its final state. -func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { +func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, + startingPkt *SidecarPacket) { + defer a.wg.Done() packetChan := make(chan *sidecar.Ticket, 1) @@ -183,9 +236,8 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { "stream_id=%x", startingPkt.ReceiverTicket.ID[:], recipientStreamID[:]) - err = a.client.InitTicketCipherBox( - context.Background(), recipientStreamID, - startingPkt.ReceiverTicket, + err = a.cfg.MailBox.InitSidecarMailbox( + recipientStreamID, startingPkt.ReceiverTicket, ) if err != nil && !isErrAlreadyExists(err) { log.Errorf("unable to init cipher box: %v", err) @@ -204,8 +256,8 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { // stream and deliver them to the main gorotuine until we // receive a message over the cancel channel. for { - newTicket, err := a.recvSidecarPkt( - startingPkt.ReceiverTicket, false, + newTicket, err := a.cfg.MailBox.RecvSidecarPkt( + ctx, startingPkt.ReceiverTicket, false, ) if err != nil { // TODO(roasbeef): back off then retry? @@ -218,7 +270,7 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { case <-cancelChan: return - case <-a.quit: + case <-a.cfg.Quit: return } @@ -229,7 +281,7 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { select { case newTicket := <-packetChan: - newPktState, err := a.stateStepRecipient(&SidecarPacket{ + newPktState, err := a.stateStepRecipient(ctx, &SidecarPacket{ CurrentState: currentState, ProviderTicket: newTicket, ReceiverTicket: localTicket, @@ -253,7 +305,7 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { return } - case <-a.quit: + case <-a.cfg.Quit: return } } @@ -263,8 +315,8 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { // receiver through the sidecar negotiation process. It takes the current state // (the state of the goroutine, and the incoming ticket) and maps that into a // new state, with a possibly modified ticket. -func (a *SidecarAcceptor) stateStepRecipient(pkt *SidecarPacket, -) (*SidecarPacket, error) { +func (a *SidecarNegotiator) stateStepRecipient(ctx context.Context, + pkt *SidecarPacket) (*SidecarPacket, error) { switch { @@ -289,7 +341,7 @@ func (a *SidecarAcceptor) stateStepRecipient(pkt *SidecarPacket, log.Infof("Transmitting registered ticket=%x to provider", pkt.ProviderTicket.ID[:]) - err := a.sendSidecarPkt(pkt.ReceiverTicket, true) + err := a.cfg.MailBox.SendSidecarPkt(ctx, pkt.ReceiverTicket, true) if err != nil { return nil, fmt.Errorf("unable to send pkt: %w", err) } @@ -312,10 +364,7 @@ func (a *SidecarAcceptor) stateStepRecipient(pkt *SidecarPacket, // At this point, we'll finish validating the ticket, then // await the ticket on the side lines if it's valid. - ctx := context.Background() - err := validateOrderedTicket( - ctx, pkt.ProviderTicket, a.cfg.Signer, a.cfg.SidecarDB, - ) + err := a.cfg.Driver.ValidateOrderedTicket(pkt.ProviderTicket) if err != nil { return nil, fmt.Errorf("unable to verify ticket: "+ "%w", err) @@ -327,14 +376,14 @@ func (a *SidecarAcceptor) stateStepRecipient(pkt *SidecarPacket, // Now that we know the channel is valid, we'll wait for the // channel to show up at our node, and allow things to advance // to the completion state. - err = a.ExpectChannel(ctx, pkt.ProviderTicket) + err = a.cfg.Driver.ExpectChannel( + ctx, pkt.ProviderTicket, + ) if err != nil { return nil, fmt.Errorf("failed to expect "+ "channel: %w", err) } - // TODO(roasbeef): set state to expecting channel? - return &SidecarPacket{ CurrentState: sidecar.StateExpectingChannel, ReceiverTicket: pkt.ProviderTicket, @@ -354,7 +403,7 @@ func (a *SidecarAcceptor) stateStepRecipient(pkt *SidecarPacket, // autoSidecarProvider is a goroutine that will attempt to advance a new // sidecar ticket through the negotiation process until it reaches its final // state. -func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, +func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt *SidecarPacket, bid *order.Bid, acct *account.Account) { defer a.wg.Done() @@ -384,9 +433,7 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, log.Infof("Creating provider mailbox for ticket=%x, w/ stream_id=%x", localTicket.ID[:], streamID[:]) - err = a.client.InitAccountCipherBox( - context.Background(), streamID, acct.TraderKey, - ) + err = a.cfg.MailBox.InitAcctMailbox(streamID, acct.TraderKey) if err != nil && !isErrAlreadyExists(err) { log.Errorf("unable to init cipher box: %v", err) return @@ -400,8 +447,8 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, // stream and deliver them to the main gorotuine until we // receive a message over the cancel channel. for { - newTicket, err := a.recvSidecarPkt( - startingPkt.ProviderTicket, true, + newTicket, err := a.cfg.MailBox.RecvSidecarPkt( + ctx, startingPkt.ProviderTicket, true, ) if err != nil { log.Error(err) @@ -413,7 +460,7 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, case <-cancelChan: return - case <-a.quit: + case <-a.cfg.Quit: return } @@ -429,7 +476,7 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, for { priorState := currentState - newPktState, err := a.stateStepProvider(&SidecarPacket{ + newPktState, err := a.stateStepProvider(ctx, &SidecarPacket{ CurrentState: currentState, ReceiverTicket: newTicket, ProviderTicket: localTicket, @@ -462,7 +509,7 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, } } - case <-a.quit: + case <-a.cfg.Quit: return } } @@ -471,8 +518,8 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, // stateStepProvider is the state transition function for the provider of a // sidecar ticket. It takes the current transcript state, the provider's // account, and canned bid and returns a new transition to a new ticket state. -func (a *SidecarAcceptor) stateStepProvider(pkt *SidecarPacket, bid *order.Bid, - acct *account.Account) (*SidecarPacket, error) { +func (a *SidecarNegotiator) stateStepProvider(ctx context.Context, + pkt *SidecarPacket, bid *order.Bid, acct *account.Account) (*SidecarPacket, error) { switch { // In this case, we've just restarted, so we'll attempt to start from @@ -485,7 +532,7 @@ func (a *SidecarAcceptor) stateStepProvider(pkt *SidecarPacket, bid *order.Bid, log.Infof("Resuming negotiation for ticket=%x, requesting "+ "registered ticket", pkt.ProviderTicket.ID[:]) - err := a.sendSidecarPkt(pkt.ProviderTicket, false) + err := a.cfg.MailBox.SendSidecarPkt(ctx, pkt.ProviderTicket, false) if err != nil { return nil, err } @@ -509,7 +556,7 @@ func (a *SidecarAcceptor) stateStepProvider(pkt *SidecarPacket, bid *order.Bid, // Now that we have the ticket, we'll update the state on disk // to checkpoint the new state. - err := a.cfg.SidecarDB.UpdateSidecar(pkt.ReceiverTicket) + err := a.cfg.Driver.UpdateSidecar(pkt.ReceiverTicket) if err != nil { return nil, fmt.Errorf("unable to update ticket: %w", err) @@ -533,8 +580,8 @@ func (a *SidecarAcceptor) stateStepProvider(pkt *SidecarPacket, bid *order.Bid, // Now we have the recipient's information, we can attach it to // our bid, and submit it as normal. - updatedTicket, err := a.submitSidecarOrder( - context.Background(), pkt.ProviderTicket, bid, acct, + updatedTicket, err := a.cfg.Driver.SubmitSidecarOrder( + pkt.ProviderTicket, bid, acct, ) switch { // If the order has already been submitted, then we'll catch @@ -578,7 +625,7 @@ func (a *SidecarAcceptor) stateStepProvider(pkt *SidecarPacket, bid *order.Bid, // we send over is in the state they expect. pkt.ProviderTicket.State = sidecar.StateOrdered - err := a.sendSidecarPkt(pkt.ProviderTicket, false) + err := a.cfg.MailBox.SendSidecarPkt(ctx, pkt.ProviderTicket, false) if err != nil { return nil, fmt.Errorf("unable to send sidecar "+ "pkt: %v", err) @@ -591,7 +638,7 @@ func (a *SidecarAcceptor) stateStepProvider(pkt *SidecarPacket, bid *order.Bid, // disk to checkpoint the new state. If the remote party ends // us any messages after we persist this state, then we'll // simply re-send the latest ticket. - err = a.cfg.SidecarDB.UpdateSidecar(&updatedTicket) + err = a.cfg.Driver.UpdateSidecar(&updatedTicket) if err != nil { return nil, fmt.Errorf("unable to update ticket: %w", err) diff --git a/sidecar_acceptor.go b/sidecar_acceptor.go index 2ce6cf1..f28ff6e 100644 --- a/sidecar_acceptor.go +++ b/sidecar_acceptor.go @@ -18,6 +18,8 @@ import ( "github.com/lightninglabs/pool/sidecar" "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/subscribe" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // SidecarAcceptor is a type that is exclusively responsible for managing the @@ -132,11 +134,20 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { // sidecar ticket. case err == clientdb.ErrAccountNotFound: - go a.autoSidecarReceiver(&SidecarPacket{ - CurrentState: ticket.State, - ReceiverTicket: ticket, - ProviderTicket: ticket, + autoAcceptor := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: false, + StartingPkt: &SidecarPacket{ + CurrentState: ticket.State, + ReceiverTicket: ticket, + ProviderTicket: ticket, + }, + Driver: a, + MailBox: a, + Quit: a.quit, }) + if err := autoAcceptor.Start(); err != nil { + return err + } // Otherwise, we're on the other end of things, so // we'll assume the role of the provider. @@ -162,12 +173,22 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { state = sidecar.StateCreated } - // TODO(roasbeef): state to cause to re-send? - go a.autoSidecarProvider(&SidecarPacket{ - CurrentState: state, - ReceiverTicket: ticket, - ProviderTicket: ticket, - }, ticketBid, acct) + autoAcceptor := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: true, + ProviderBid: ticketBid, + StartingPkt: &SidecarPacket{ + CurrentState: state, + ReceiverTicket: ticket, + ProviderTicket: ticket, + }, + ProviderAccount: acct, + Driver: a, + MailBox: a, + Quit: a.quit, + }) + if err := autoAcceptor.Start(); err != nil { + return err + } default: return fmt.Errorf("unable to fetch account "+ @@ -376,30 +397,32 @@ func (a *SidecarAcceptor) AutoAcceptSidecar(ticket *sidecar.Ticket) error { log.Infof("Attempting negotiation to receive sidecar ticket: %x", ticket.ID[:]) - // We'll launch a new coroutine that'll handle negotiation in the - // background all the way to the final state of the ticket. - a.wg.Add(1) - go a.autoSidecarReceiver(&SidecarPacket{ - CurrentState: sidecar.StateRegistered, - ProviderTicket: ticket, - ReceiverTicket: ticket, + autoAcceptor := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: false, + StartingPkt: &SidecarPacket{ + CurrentState: sidecar.StateRegistered, + ReceiverTicket: ticket, + ProviderTicket: ticket, + }, + Driver: a, + MailBox: a, + Quit: a.quit, }) - - return nil + return autoAcceptor.Start() } -// submitSidecarOrder attempts to submit a new bid that's bound to a finalized +// SubmitSidecarOrder attempts to submit a new bid that's bound to a finalized // sidecar ticket that's in the registered phase. If this method returns // successfully, then the ticket will have transitioned to the // sidecar.StateOrdered state. -func (a *SidecarAcceptor) submitSidecarOrder(ctx context.Context, - ticket *sidecar.Ticket, bid *order.Bid, +func (a *SidecarAcceptor) SubmitSidecarOrder(ticket *sidecar.Ticket, bid *order.Bid, acct *account.Account) (*sidecar.Ticket, error) { // We'll bind the ticket to the order now as the ticket has all the // necessary information included. bid.SidecarTicket = ticket + ctx := context.Background() auctionTerms, err := a.client.Terms(ctx) if err != nil { return nil, fmt.Errorf("could not query auctioneer terms: %v", err) @@ -424,14 +447,20 @@ func (a *SidecarAcceptor) CoordinateSidecar(ticket *sidecar.Ticket, log.Infof("Attempting negotiation to offer sidecar ticket: %x", ticket.ID[:]) - a.wg.Add(1) - go a.autoSidecarProvider(&SidecarPacket{ - CurrentState: sidecar.StateOffered, - ProviderTicket: ticket, - ReceiverTicket: ticket, - }, bid, acct) - - return nil + autoAcceptor := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: true, + ProviderBid: bid, + StartingPkt: &SidecarPacket{ + CurrentState: sidecar.StateOffered, + ProviderTicket: ticket, + ReceiverTicket: ticket, + }, + ProviderAccount: acct, + Driver: a, + MailBox: a, + Quit: a.quit, + }) + return autoAcceptor.Start() } // handleServerMessage reacts to a message sent by the server and sends back the @@ -688,3 +717,114 @@ func (a *SidecarAcceptor) removeShims(batch *order.Batch) error { return nil } + +// UpdateSidecar writes the passed sidecar ticket to persistent storage. +func (a *SidecarAcceptor) UpdateSidecar(tkt *sidecar.Ticket) error { + return a.cfg.SidecarDB.UpdateSidecar(tkt) +} + +// ValidateOrderedTicketctx attempts to validate that a given ticket has +// properly transitioned to the ordered state. +func (a *SidecarAcceptor) ValidateOrderedTicket(tkt *sidecar.Ticket) error { + ctx := context.Background() + return validateOrderedTicket(ctx, tkt, a.cfg.Signer, a.cfg.SidecarDB) +} + +// InitAcctMailbox attempts to create the mailbox with the given stream ID +// using account signature authentication mechanism. If the mailbox already +// exists, then a nil error is to be returned. +func (a *SidecarAcceptor) InitAcctMailbox(streamID [64]byte, + traderKey *keychain.KeyDescriptor) error { + + err := a.client.InitAccountCipherBox( + context.Background(), streamID, traderKey, + ) + if err != nil && !isErrAlreadyExists(err) { + return fmt.Errorf("unable to init cipher box: %v", err) + + } + + return nil +} + +// InitSidecarMailbox attempts to create the mailbox with the given stream ID +// using the sidecar ticket authentication mechanism. If the mailbox already +// exists, then a nil error is to be returned. +func (a *SidecarAcceptor) InitSidecarMailbox(streamID [64]byte, + tkt *sidecar.Ticket) error { + + err := a.client.InitTicketCipherBox(context.Background(), streamID, tkt) + if err != nil && !isErrAlreadyExists(err) { + return fmt.Errorf("unable to init cipher box: %v", err) + } + + return nil +} + +// SendSidecarPkt attempts to send a sidecar packet to the opposite party using +// their registered cipherbox stream. +func (a *SidecarAcceptor) SendSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, + provider bool) error { + + var ticketBuf bytes.Buffer + err := sidecar.SerializeTicket(&ticketBuf, pkt) + if err != nil { + return err + } + + streamID, err := deriveStreamID(pkt, provider) + if err != nil { + return err + } + + target := "receiver" + if provider { + target = "provider" + } + + log.Infof("Sending ticket(state=%v, id=%x) to %v stream_id=%x", + pkt.State, pkt.ID[:], target, streamID[:]) + + return a.client.SendCipherBoxMsg(ctx, streamID, ticketBuf.Bytes()) +} + +// RecvSidecarPkt attempts to receive a new sidecar packet from the opposite +// party using their registered cipherbox stream. +func (a *SidecarAcceptor) RecvSidecarPkt(pCtx context.Context, + ticket *sidecar.Ticket, provider bool) (*sidecar.Ticket, error) { + + streamID, err := deriveStreamID(ticket, provider) + if err != nil { + return nil, err + } + + log.Infof("Waiting for ticket (id=%x) using stream_id=%x, provider=%v", + ticket.ID[:], streamID[:], provider) + + ctx, cancel := context.WithCancel(pCtx) + defer cancel() + + msg, err := a.client.RecvCipherBoxMsg(ctx, streamID) + if err != nil { + return nil, fmt.Errorf("unable to recv cipher box "+ + "msg: %w", err) + } + + log.Infof("Receive new message for ticket (id=%x) "+ + "via stream_id=%x, provider=%v", ticket.ID[:], streamID, + provider) + + return sidecar.DeserializeTicket(bytes.NewReader(msg)) +} + +// isErrAlreadyExists returns true if the passed error is the "already exists" +// error within the error wrapped error which is returned by the hash mail +// server when a stream we're attempting to create already exists. +func isErrAlreadyExists(err error) bool { + statusCode, ok := status.FromError(err) + if !ok { + return false + } + + return statusCode.Code() == codes.AlreadyExists +} From 3369357053ca0350a1385c7d126036a2699a8c89 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:09:18 -0700 Subject: [PATCH 05/12] auctioneer: add client methods for HashMail stream deletion --- auctioneer/client.go | 88 ++++++++++++++++++++++++++++++++++---------- 1 file changed, 69 insertions(+), 19 deletions(-) diff --git a/auctioneer/client.go b/auctioneer/client.go index 3ef7b73..655540b 100644 --- a/auctioneer/client.go +++ b/auctioneer/client.go @@ -1321,20 +1321,17 @@ func (c *Client) MarketInfo(ctx context.Context) ( return c.client.MarketInfo(ctx, &auctioneerrpc.MarketInfoRequest{}) } -// InitAccountCipherBox attempts to initialize a new CipherBox using the -// sidecar ticket as the authentication method. -func (c *Client) InitTicketCipherBox(ctx context.Context, sid [64]byte, - ticket *sidecar.Ticket) error { - - // TODO(roasbeef): add error to catch deupliacte stream - // existence/creation, also need to allow stream deletion as well +// genSidecarAuth generates a set of valid authentication details to allow +// creating or deleting a hashmail mailbox. +func genSidecarAuth(sid [64]byte, + ticket *sidecar.Ticket) (*auctioneerrpc.CipherBoxAuth, error) { strTicket, err := sidecar.EncodeToString(ticket) if err != nil { - return err + return nil, err } - streamInit := &auctioneerrpc.CipherBoxAuth{ + return &auctioneerrpc.CipherBoxAuth{ Desc: &auctioneerrpc.CipherBoxDesc{ StreamId: sid[:], }, @@ -1343,25 +1340,52 @@ func (c *Client) InitTicketCipherBox(ctx context.Context, sid [64]byte, Ticket: strTicket, }, }, - } - _, err = c.hashMailClient.NewCipherBox(ctx, streamInit) - return err + }, nil } // InitAccountCipherBox attempts to initialize a new CipherBox using the -// account key as an authentication mechanism. -func (c *Client) InitAccountCipherBox(ctx context.Context, sid [64]byte, - acctKey *keychain.KeyDescriptor) error { +// sidecar ticket as the authentication method. +func (c *Client) InitTicketCipherBox(ctx context.Context, sid [64]byte, + ticket *sidecar.Ticket) error { - streamSig, err := c.cfg.Signer.SignMessage( + streamAuth, err := genSidecarAuth(sid, ticket) + if err != nil { + return err + } + + _, err = c.hashMailClient.NewCipherBox(ctx, streamAuth) + return err +} + +// DelSidecarMailbox tears down the mailbox the sidecar ticket recipient used +// to communicate with the provider. +func (c *Client) DelSidecarMailbox(ctx context.Context, + streamID [64]byte, ticket *sidecar.Ticket) error { + + streamAuth, err := genSidecarAuth(streamID, ticket) + if err != nil { + return err + } + + _, err = c.hashMailClient.DelCipherBox(ctx, streamAuth) + return err +} + +// genAcctAuth generates a valid authentication sig to allow a trader to delete +// or create a new hashmail mailbox. +func genAcctAuth(ctx context.Context, signer lndclient.SignerClient, + sid [64]byte, acctKey *keychain.KeyDescriptor) (*auctioneerrpc.CipherBoxAuth, error) { + + streamSig, err := signer.SignMessage( ctx, sid[:], acctKey.KeyLocator, ) if err != nil { - return fmt.Errorf("unable to sign cipher box auth: %w", err) + return nil, fmt.Errorf("unable to sign cipher box "+ + "auth: %w", err) } acctKeyBytes := acctKey.PubKey.SerializeCompressed() - streamInit := &auctioneerrpc.CipherBoxAuth{ + return &auctioneerrpc.CipherBoxAuth{ Desc: &auctioneerrpc.CipherBoxDesc{ StreamId: sid[:], }, @@ -1371,8 +1395,34 @@ func (c *Client) InitAccountCipherBox(ctx context.Context, sid [64]byte, StreamSig: streamSig, }, }, + }, nil +} + +// InitAccountCipherBox attempts to initialize a new CipherBox using the +// account key as an authentication mechanism. +func (c *Client) InitAccountCipherBox(ctx context.Context, sid [64]byte, + acctKey *keychain.KeyDescriptor) error { + + streamAuth, err := genAcctAuth(ctx, c.cfg.Signer, sid, acctKey) + if err != nil { + return err } - _, err = c.hashMailClient.NewCipherBox(ctx, streamInit) + + _, err = c.hashMailClient.NewCipherBox(ctx, streamAuth) + return err +} + +// DelAcctMailbox tears down the mailbox that the sidecar ticket provider used +// to communicate with the recipient. +func (c *Client) DelAcctMailbox(ctx context.Context, sid [64]byte, + acctKey *keychain.KeyDescriptor) error { + + streamAuth, err := genAcctAuth(ctx, c.cfg.Signer, sid, acctKey) + if err != nil { + return err + } + + _, err = c.hashMailClient.DelCipherBox(ctx, streamAuth) return err } From 36dbabd7142aeeb30178501bb15df5c0d3ff1aea Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:10:23 -0700 Subject: [PATCH 06/12] sidecar: add CurrentState() method to side car negotiator This methods makes writing tests easier as the test context can introspect into the negotiator's state. --- auto_sidecar.go | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index a6ed2e7..4d47065 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -163,6 +163,8 @@ type AutoAcceptorConfig struct { // provider and recipient of a sidecar channel to complete the manual steps in // automated manner. type SidecarNegotiator struct { + currentState uint32 + cfg AutoAcceptorConfig wg sync.WaitGroup @@ -172,7 +174,8 @@ type SidecarNegotiator struct { // a valid config. func NewSidecarNegotiator(cfg AutoAcceptorConfig) *SidecarNegotiator { return &SidecarNegotiator{ - cfg: cfg, + cfg: cfg, + currentState: uint32(cfg.StartingPkt.CurrentState), } } @@ -291,7 +294,12 @@ func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, continue } - currentState = newPktState.CurrentState + // TODO(roasbeef): make into method for easier + // assertions? + atomic.StoreUint32( + &a.currentState, uint32(newPktState.CurrentState), + ) + localTicket = newPktState.ReceiverTicket // If our next target state is the completion state, @@ -489,6 +497,10 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt currentState = newPktState.CurrentState localTicket = newPktState.ProviderTicket + atomic.StoreUint32( + &a.currentState, uint32(newPktState.CurrentState), + ) + switch { case priorState == currentState: fallthrough @@ -515,6 +527,12 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt } } +// CurrentState returns the current state of the sidecar negotiator. +func (a *SidecarNegotiator) CurrentState() sidecar.State { + state := atomic.LoadUint32(&a.currentState) + return sidecar.State(state) +} + // stateStepProvider is the state transition function for the provider of a // sidecar ticket. It takes the current transcript state, the provider's // account, and canned bid and returns a new transition to a new ticket state. From db646c5a490bb869a25b6f4bd8bd3a210a52a7c6 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:11:44 -0700 Subject: [PATCH 07/12] sidecar: only simulate retransmission for provider if restarting from scratch We only need to re-send the simulated retransmission message if we're starting from scratch and haven't yet received the provider's message. Otherwise, this will cause an unnecessary internal state transition. --- auto_sidecar.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index 4d47065..f4e51c1 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -426,8 +426,11 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt localTicket := startingPkt.ProviderTicket // We'll start with a simulated starting message from the sidecar - // receiver. - packetChan <- startingPkt.ReceiverTicket + // receiver, but only if we're starting in the created state which + // demands an internal retransmission. + if startingPkt.CurrentState == sidecar.StateCreated { + packetChan <- startingPkt.ReceiverTicket + } // First, we'll need to derive the stream ID that we'll use to receive // new messages from the recipient. From 026addd7cdb7ecaaf638709caa1a7e0a77683771 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:12:08 -0700 Subject: [PATCH 08/12] sidecar: add new method to allow negotiator to finalize ticket state --- auto_sidecar.go | 112 +++++++++++++++++++++++++++++++++++------------- 1 file changed, 83 insertions(+), 29 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index f4e51c1..33cecd4 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "github.com/lightninglabs/pool/account" "github.com/lightninglabs/pool/clientdb" @@ -34,10 +35,18 @@ type MailBox interface { // mailbox already exists, then a nil error is to be returned. InitSidecarMailbox(streamID [64]byte, ticket *sidecar.Ticket) error + // DelSidecarMailbox tears down the mailbox the sidecar ticket + // recipient used to communicate with the provider. + DelSidecarMailbox(streamID [64]byte, ticket *sidecar.Ticket) error + // InitAcctMailbox attempts to create the mailbox with the given stream // ID using account signature authentication mechanism. If the mailbox // already exists, then a nil error is to be returned. InitAcctMailbox(streamID [64]byte, pubKey *keychain.KeyDescriptor) error + + // DelAcctMailbox tears down the mailbox that the sidecar ticket + // provider used to communicate with the recipient. + DelAcctMailbox(streamID [64]byte, pubKey *keychain.KeyDescriptor) error } // SidecarPacket encapsulates the current state of an auto sidecar negotiator. @@ -153,10 +162,6 @@ type AutoAcceptorConfig struct { // MailBox is used to allow negotiators to send messages back and forth // to each other. MailBox MailBox - - // Quit is a global quit channel that all spawned goroutines will - // select on select on. - Quit chan struct{} } // SidecarNegotiator is a sub-system that uses a mailbox abstraction between a @@ -168,6 +173,11 @@ type SidecarNegotiator struct { cfg AutoAcceptorConfig wg sync.WaitGroup + + ticketFinalized chan struct{} + quit chan struct{} + + stopOnce sync.Once } // NewSidecarNegotiator returns a new instance of the sidecar negotiator given @@ -176,13 +186,14 @@ func NewSidecarNegotiator(cfg AutoAcceptorConfig) *SidecarNegotiator { return &SidecarNegotiator{ cfg: cfg, currentState: uint32(cfg.StartingPkt.CurrentState), + ticketFinalized: make(chan struct{}), + quit: make(chan struct{}), } } // Start kicks off the set of goroutines needed for the sidecar channel to be // negotiated. func (a *SidecarNegotiator) Start() error { - // In order to ensure we can exit properly if signalled, we'll launch a // goroutine that will cancel a global context if we need to exit. a.wg.Add(1) @@ -190,7 +201,7 @@ func (a *SidecarNegotiator) Start() error { go func() { defer a.wg.Done() - <-a.cfg.Quit + <-a.quit cancel() }() @@ -208,6 +219,26 @@ func (a *SidecarNegotiator) Start() error { return nil } +// Stop signals all goroutines to enter a graceful shutdown. +func (a *SidecarNegotiator) Stop() { + a.stopOnce.Do(func() { + close(a.quit) + a.wg.Wait() + }) +} + +// TicketExecuted is a clean up function that should be called once the ticket +// has been executeed, meaning a channel defined by it was confirmed ina batch +// on chain. +func (a *SidecarNegotiator) TicketExecuted() { + select { + case a.ticketFinalized <- struct{}{}: + case <-a.quit: + } + + a.Stop() +} + // autoSidecarReceiver is a goroutine that will attempt to advance a new // sidecar ticket through the process until it reaches its final state. func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, @@ -273,7 +304,7 @@ func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, case <-cancelChan: return - case <-a.cfg.Quit: + case <-a.quit: return } @@ -302,18 +333,30 @@ func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, localTicket = newPktState.ReceiverTicket - // If our next target state is the completion state, - // then our job here is done, and we can safely exit - // this main goroutine. - if newPktState.CurrentState == sidecar.StateCompleted { - log.Infof("Receiver negotiation for " + - "SidecarTicket(%x) complete!") + case <-a.ticketFinalized: + log.Infof("Receiver negotiation for "+ + "SidecarTicket(%x) complete!", localTicket.ID[:]) - close(cancelChan) + // The ticket has been marked as finalized, so we'll + // update it as being completed in the database. + localTicket.State = sidecar.StateCompleted + if err := a.cfg.Driver.UpdateSidecar(localTicket); err != nil { + log.Errorf("unable to update ticket to "+ + "complete state: %v", err) return } - case <-a.cfg.Quit: + // We'll also tear down the mailbox as well as we no + // longer need it anymore. + err := a.cfg.MailBox.DelSidecarMailbox( + recipientStreamID, localTicket, + ) + if err != nil { + log.Errorf("unable to reclaim mailbox: %v", err) + return + } + + case <-a.quit: return } } @@ -471,7 +514,7 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt case <-cancelChan: return - case <-a.cfg.Quit: + case <-a.quit: return } @@ -509,22 +552,33 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt fallthrough case currentState == sidecar.StateExpectingChannel: break - - // If our next target state is the completion - // state, then our job here is done, and we can - // safely exit this main goroutine. - case newPktState.CurrentState == - sidecar.StateCompleted: - - log.Infof("Receiver negotiation for " + - "SidecarTicket(%x) complete!") - - close(cancelChan) - return } } - case <-a.cfg.Quit: + case <-a.ticketFinalized: + log.Infof("Receiver negotiation for SidecarTicket(%x) "+ + "complete!", localTicket.ID[:]) + + // The ticket has been marked as finalized, so we'll + // update it as being completed in the database. + localTicket.State = sidecar.StateCompleted + if err := a.cfg.Driver.UpdateSidecar(localTicket); err != nil { + log.Errorf("unable to update ticket to "+ + "complete state: %v", err) + return + } + + // We'll also tear down the mailbox as well as we no + // longer need it anymore. + err := a.cfg.MailBox.DelAcctMailbox( + streamID, acct.TraderKey, + ) + if err != nil { + log.Errorf("unable to reclaim mailbox: %v", err) + return + } + + case <-a.quit: return } } From 3b4a2cdcb97f771603b20431c186df909bfb1e94 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:13:07 -0700 Subject: [PATCH 09/12] sidecar: maintain set of negotiators to ensure graceful shutdown In this commit, we being to collect a map of all the negotiators so we can ensure that if the main daemon needs to shutdown, then they do as well. --- auto_sidecar.go | 5 +++ sidecar_acceptor.go | 76 +++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index 33cecd4..aafd1ee 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -16,6 +16,11 @@ import ( "github.com/lightningnetwork/lnd/lnwire" ) +// TODO(roasbeef): need to ensure have updates for termination, etc +// * finalize for the recreiver +// * need other hook for the provider +// * also delete as well (the mailbox) + // MailBox is an interface that abstracts over the HashMail functionality to // represent a generic mailbox that both sides will use to communicate with // each other. diff --git a/sidecar_acceptor.go b/sidecar_acceptor.go index f28ff6e..e360e6b 100644 --- a/sidecar_acceptor.go +++ b/sidecar_acceptor.go @@ -47,12 +47,18 @@ type SidecarAcceptor struct { sync.Mutex + // negotiators maps a potential stream ID (of the recipient) to the + // active sidecar negotiator. We'll maintain this map to be able to + // shutdown the negotiators, as well as notify them that the ticket has + // been fully executed. + negotiators map[[64]byte]*SidecarNegotiator + quit chan struct{} wg sync.WaitGroup } // SidecarAcceptorConfig holds all the configuration information that sidecar -// acceptor needs in order to carry out its dutes. +// acceptor needs in order to carry out its duties type SidecarAcceptorConfig struct { SidecarDB sidecar.Store @@ -86,6 +92,7 @@ func NewSidecarAcceptor(cfg *SidecarAcceptorConfig) *SidecarAcceptor { cfg: cfg, pendingSidecarOrders: make(map[order.Nonce]*sidecar.Ticket), quit: make(chan struct{}), + negotiators: make(map[[64]byte]*SidecarNegotiator), } } @@ -149,6 +156,16 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { return err } + streamID, err := deriveRecipientStreamID(ticket) + if err != nil { + return fmt.Errorf("unable to derive "+ + "stream IDs: %v", err) + } + + a.Lock() + a.negotiators[streamID] = autoAcceptor + a.Unlock() + // Otherwise, we're on the other end of things, so // we'll assume the role of the provider. case err == nil: @@ -190,6 +207,16 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { return err } + streamID, err := deriveRecipientStreamID(ticket) + if err != nil { + return fmt.Errorf("unable to derive "+ + "stream IDs: %v", err) + } + + a.Lock() + a.negotiators[streamID] = autoAcceptor + a.Unlock() + default: return fmt.Errorf("unable to fetch account "+ "for sidecar: %w", err) @@ -261,6 +288,10 @@ func (a *SidecarAcceptor) Stop() error { returnErr = err } + for _, negotiator := range a.negotiators { + negotiator.Stop() + } + a.pendingOpenChanClient.Cancel() a.cfg.Acceptor.Stop() close(a.quit) @@ -408,6 +439,17 @@ func (a *SidecarAcceptor) AutoAcceptSidecar(ticket *sidecar.Ticket) error { MailBox: a, Quit: a.quit, }) + + streamID, err := deriveRecipientStreamID(ticket) + if err != nil { + return fmt.Errorf("unable to derive "+ + "stream IDs: %v", err) + } + + a.Lock() + a.negotiators[streamID] = autoAcceptor + a.Unlock() + return autoAcceptor.Start() } @@ -458,8 +500,18 @@ func (a *SidecarAcceptor) CoordinateSidecar(ticket *sidecar.Ticket, ProviderAccount: acct, Driver: a, MailBox: a, - Quit: a.quit, }) + + streamID, err := deriveRecipientStreamID(ticket) + if err != nil { + return fmt.Errorf("unable to derive "+ + "stream IDs: %v", err) + } + + a.Lock() + a.negotiators[streamID] = autoAcceptor + a.Unlock() + return autoAcceptor.Start() } @@ -817,6 +869,26 @@ func (a *SidecarAcceptor) RecvSidecarPkt(pCtx context.Context, return sidecar.DeserializeTicket(bytes.NewReader(msg)) } +// DelSidecarMailbox tears down the mailbox the sidecar ticket recipient used +// to communicate with the provider. +func (a *SidecarAcceptor) DelSidecarMailbox(streamID [64]byte, + ticket *sidecar.Ticket) error { + + return a.client.DelSidecarMailbox( + context.Background(), streamID, ticket, + ) +} + +// DelAcctMailbox tears down the mailbox that the sidecar ticket provider used +// to communicate with the recipient. +func (a *SidecarAcceptor) DelAcctMailbox(streamID [64]byte, + pubKey *keychain.KeyDescriptor) error { + + return a.client.DelAcctMailbox( + context.Background(), streamID, pubKey, + ) +} + // isErrAlreadyExists returns true if the passed error is the "already exists" // error within the error wrapped error which is returned by the hash mail // server when a stream we're attempting to create already exists. From c1c67195eda7daf41b772fdf052f900c74693541 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:13:27 -0700 Subject: [PATCH 10/12] sidecar: shutdown receiver's negotiator after batch finalize --- sidecar_acceptor.go | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/sidecar_acceptor.go b/sidecar_acceptor.go index e360e6b..9e75b80 100644 --- a/sidecar_acceptor.go +++ b/sidecar_acceptor.go @@ -150,7 +150,6 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { }, Driver: a, MailBox: a, - Quit: a.quit, }) if err := autoAcceptor.Start(); err != nil { return err @@ -201,7 +200,6 @@ func (a *SidecarAcceptor) Start(errChan chan error) error { ProviderAccount: acct, Driver: a, MailBox: a, - Quit: a.quit, }) if err := autoAcceptor.Start(); err != nil { return err @@ -437,7 +435,6 @@ func (a *SidecarAcceptor) AutoAcceptSidecar(ticket *sidecar.Ticket) error { }, Driver: a, MailBox: a, - Quit: a.quit, }) streamID, err := deriveRecipientStreamID(ticket) @@ -703,9 +700,25 @@ func (a *SidecarAcceptor) matchFinalize(batch *order.Batch) { delete(a.pendingSidecarOrders, ourOrder) a.pendingSidecarOrdersMtx.Unlock() - // TODO(roasbeef): send message to the other goroutine here as well - a.cfg.Acceptor.ShimRemoved(dummyBid.(*order.Bid)) + + streamID, err := deriveRecipientStreamID(ticket) + if err != nil { + log.Errorf("unable to derive stream IDs: %v", err) + } + + // We'll also signal to the negotiator (if it exists) that the + // ticket has been finalized so it can safely exit. We don't + // need to hold the main lock here as handleServerMessage + // obtains the lock while these methods are called. + negotiator, ok := a.negotiators[streamID] + if !ok { + return + } + + negotiator.TicketExecuted() + + delete(a.negotiators, streamID) } } From c8469eb7db81b08666e7b61b8980e1c6db88caf9 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 21 May 2021 20:13:47 -0700 Subject: [PATCH 11/12] sidecar: add unit tests for basic ticket negotiation and restart --- sidecar_acceptor_test.go | 418 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 418 insertions(+) diff --git a/sidecar_acceptor_test.go b/sidecar_acceptor_test.go index e933d50..2f9f997 100644 --- a/sidecar_acceptor_test.go +++ b/sidecar_acceptor_test.go @@ -2,16 +2,22 @@ package pool import ( "context" + "fmt" "io/ioutil" "math/big" "os" "testing" + "time" "github.com/btcsuite/btcd/btcec" + "github.com/lightninglabs/pool/account" "github.com/lightninglabs/pool/auctioneer" "github.com/lightninglabs/pool/clientdb" "github.com/lightninglabs/pool/internal/test" + "github.com/lightninglabs/pool/order" "github.com/lightninglabs/pool/sidecar" + "github.com/lightningnetwork/lnd/keychain" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -156,3 +162,415 @@ func TestRegisterSidecar(t *testing.T) { cleanup() } } + +type mockMailBox struct { + providerChan chan *sidecar.Ticket + providerMsgAck chan struct{} + providerDel chan struct{} + + receiverChan chan *sidecar.Ticket + receiverMsgAck chan struct{} + receiverDel chan struct{} +} + +func newMockMailBox() *mockMailBox { + return &mockMailBox{ + providerChan: make(chan *sidecar.Ticket), + providerMsgAck: make(chan struct{}), + providerDel: make(chan struct{}), + + receiverChan: make(chan *sidecar.Ticket), + receiverMsgAck: make(chan struct{}), + receiverDel: make(chan struct{}), + } +} + +func (m *mockMailBox) RecvSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, + provider bool) (*sidecar.Ticket, error) { + + var ( + recvChan chan *sidecar.Ticket + ackChan chan struct{} + ) + if provider { + recvChan = m.providerChan + ackChan = m.providerMsgAck + } else { + recvChan = m.receiverChan + ackChan = m.receiverMsgAck + } + + select { + case <-ctx.Done(): + return nil, fmt.Errorf("mailbox shutting down") + + case tkt := <-recvChan: + + ackChan <- struct{}{} + + return tkt, nil + } +} + +func (m *mockMailBox) SendSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, + provider bool) error { + var sendChan chan *sidecar.Ticket + if provider { + sendChan = m.providerChan + } else { + sendChan = m.receiverChan + } + + select { + case <-ctx.Done(): + case sendChan <- pkt: + } + + return nil +} + +func (m *mockMailBox) InitSidecarMailbox(streamID [64]byte, ticket *sidecar.Ticket) error { + return nil +} + +func (m *mockMailBox) InitAcctMailbox(streamID [64]byte, pubKey *keychain.KeyDescriptor) error { + return nil +} + +func (m *mockMailBox) DelSidecarMailbox(streamID [64]byte, ticket *sidecar.Ticket) error { + m.receiverDel <- struct{}{} + return nil +} + +func (m *mockMailBox) DelAcctMailbox(streamID [64]byte, pubKey *keychain.KeyDescriptor) error { + m.providerDel <- struct{}{} + return nil +} + +type mockDriver struct { + stateUpdates chan sidecar.State + bidSubmitted chan struct{} + ticketValidated chan struct{} + channelExpected chan struct{} +} + +func newMockDriver() *mockDriver { + return &mockDriver{ + stateUpdates: make(chan sidecar.State), + bidSubmitted: make(chan struct{}), + ticketValidated: make(chan struct{}), + channelExpected: make(chan struct{}), + } +} + +func (m *mockDriver) ValidateOrderedTicket(tkt *sidecar.Ticket) error { + if tkt.State != sidecar.StateOrdered { + return fmt.Errorf("sidecar not in state ordered: %v", tkt.State) + } + + m.ticketValidated <- struct{}{} + + return nil +} + +func (m *mockDriver) ExpectChannel(ctx context.Context, tkt *sidecar.Ticket) error { + tkt.State = sidecar.StateExpectingChannel + + m.channelExpected <- struct{}{} + + return nil +} + +func (m *mockDriver) UpdateSidecar(tkt *sidecar.Ticket) error { + m.stateUpdates <- tkt.State + + return nil +} + +func (m *mockDriver) SubmitSidecarOrder(tkt *sidecar.Ticket, bid *order.Bid, + acct *account.Account) (*sidecar.Ticket, error) { + + tkt.State = sidecar.StateOrdered + + m.bidSubmitted <- struct{}{} + + return tkt, nil +} + +type sidecarTestCtx struct { + t *testing.T + + provider *SidecarNegotiator + providerDriver *mockDriver + + recipient *SidecarNegotiator + recipientDriver *mockDriver + + mailbox *mockMailBox +} + +func (s *sidecarTestCtx) startNegotiators() error { + if err := s.provider.Start(); err != nil { + return err + } + + return s.recipient.Start() +} + +func (s *sidecarTestCtx) restartAllNegotiators() error { + s.provider.Stop() + s.recipient.Stop() + + s.provider.quit = make(chan struct{}) + s.recipient.quit = make(chan struct{}) + + if err := s.provider.Start(); err != nil { + return err + } + + return s.recipient.Start() +} + +func (s *sidecarTestCtx) assertProviderMsgRecv() { + select { + case <-s.mailbox.providerMsgAck: + case <-time.After(time.Second * 5): + } +} + +func (s *sidecarTestCtx) assertRecipientMsgRecv() { + select { + case <-s.mailbox.receiverMsgAck: + case <-time.After(time.Second * 5): + } +} + +func (s *sidecarTestCtx) assertProviderTicketUpdated(expectedState sidecar.State) { + select { + case stateUpdate := <-s.providerDriver.stateUpdates: + + if stateUpdate != expectedState { + s.t.Fatalf("expected state=%v, got: %v", expectedState, + stateUpdate) + } + + case <-time.After(time.Second * 5): + s.t.Fatalf("provider ticket never updated") + } +} + +func (s *sidecarTestCtx) assertBidSubmited() { + select { + case <-s.providerDriver.bidSubmitted: + case <-time.After(time.Second * 5): + s.t.Fatalf("provider bid never submitted") + } +} + +func (s *sidecarTestCtx) assertRecipientTicketValidated() { + select { + case <-s.recipientDriver.ticketValidated: + case <-time.After(time.Second * 5): + s.t.Fatalf("recipient ticket never validated") + } +} + +func (s *sidecarTestCtx) assertRecipientExpectsChannel() { + select { + case <-s.recipientDriver.channelExpected: + case <-time.After(time.Second * 5): + s.t.Fatalf("recipient channel never expected") + } +} + +func (s *sidecarTestCtx) assertRecipientTicketUpdated(expectedState sidecar.State) { + select { + case stateUpdate := <-s.recipientDriver.stateUpdates: + + if stateUpdate != expectedState { + s.t.Fatalf("expected state=%v, got: %v", expectedState, + stateUpdate) + } + + case <-time.After(time.Second * 5): + s.t.Fatalf("recipient ticket never updated") + } +} + +func (s *sidecarTestCtx) confirmSidecarBatch() { + s.provider.TicketExecuted() + + s.recipient.TicketExecuted() +} + +func (s *sidecarTestCtx) assertProviderMailboxDel() { + select { + case <-s.mailbox.providerDel: + case <-time.After(time.Second * 5): + s.t.Fatalf("provider mailbox not deleted") + } +} + +func (s *sidecarTestCtx) assertRecipientMailboxDel() { + select { + case <-s.mailbox.receiverDel: + case <-time.After(time.Second * 5): + s.t.Fatalf("provider mailbox not deleted") + } +} + +func newSidecarTestCtx(t *testing.T) *sidecarTestCtx { + mailBox := newMockMailBox() + providerDriver := newMockDriver() + recipientDriver := newMockDriver() + + ticketID := [8]byte{1} + + provider := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: true, + ProviderBid: &order.Bid{ + Kit: order.Kit{ + Version: order.VersionSelfChanBalance, + LeaseDuration: 144, + MaxBatchFeeRate: 253, + MinUnitsMatch: 1, + Amt: 0, + UnitsUnfulfilled: 0, + }, + SelfChanBalance: 1, + }, + ProviderAccount: &account.Account{}, + StartingPkt: &SidecarPacket{ + CurrentState: sidecar.StateOffered, + ProviderTicket: &sidecar.Ticket{ + ID: ticketID, + State: sidecar.StateOffered, + Offer: sidecar.Offer{ + SigOfferDigest: testOfferSig, + }, + }, + ReceiverTicket: &sidecar.Ticket{ + ID: ticketID, + State: sidecar.StateOffered, + Offer: sidecar.Offer{ + SigOfferDigest: testOfferSig, + }, + }, + }, + Driver: providerDriver, + MailBox: mailBox, + }) + + recipient := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: false, + StartingPkt: &SidecarPacket{ + CurrentState: sidecar.StateRegistered, + ProviderTicket: &sidecar.Ticket{ + ID: ticketID, + State: sidecar.StateRegistered, + Offer: sidecar.Offer{ + SigOfferDigest: testOfferSig, + }, + }, + ReceiverTicket: &sidecar.Ticket{ + ID: ticketID, + State: sidecar.StateRegistered, + Offer: sidecar.Offer{ + SigOfferDigest: testOfferSig, + }, + }, + }, + Driver: recipientDriver, + MailBox: mailBox, + }) + + return &sidecarTestCtx{ + t: t, + provider: provider, + providerDriver: providerDriver, + recipient: recipient, + recipientDriver: recipientDriver, + mailbox: mailBox, + } +} + +// TestAutoSidecarNegotiation tests the routine sidecar negotiation process +// including that both sides are able to properly handle retransmissions and +// also restarts assuming persistent storage is durable. +func TestAutoSidecarNegotiation(t *testing.T) { + t.Parallel() + + testCtx := newSidecarTestCtx(t) + + // First, we'll start both negotiators. The provider should no-op, but + // then the receiver should send over the ticket and complete + // execution. At the end of the exchange, we expect that both sides are + // waiting for the channel in its expectation state. + err := testCtx.startNegotiators() + assert.NoError(t, err, fmt.Errorf("unable to start negotiators: %v", err)) + + // The recipient should send a new message to the provider with their + // ticket in the registered state. + testCtx.assertProviderMsgRecv() + + // Upon receiving the new ticket, the provider should write the new + // registered state to disk, submit the bid, then then the new ticket + // over to the recipient. + testCtx.assertProviderTicketUpdated(sidecar.StateRegistered) + testCtx.assertBidSubmited() + testCtx.assertRecipientMsgRecv() + + // After sending the ticket, the provider should update the ticket in + // its database as it waits in the expected state. + testCtx.assertProviderTicketUpdated(sidecar.StateExpectingChannel) + + // The recipient, should now validate the ticket, then wait and expect + // the channel. + testCtx.assertRecipientTicketValidated() + testCtx.assertRecipientExpectsChannel() + + // At this point, both sides should be waiting for the channel in its + // expected state. + // + // TODO(roasbeef): wrap in wait predicate? + assert.Equal( + t, sidecar.StateExpectingChannel, testCtx.provider.CurrentState(), + ) + assert.Equal( + t, sidecar.StateExpectingChannel, testCtx.recipient.CurrentState(), + ) + + // We'll now simulate a restart on both sides by signalling their + // goroutines to exit, then re-starting them anew with their persisted + // state. + err = testCtx.restartAllNegotiators() + assert.NoError(t, err, fmt.Errorf("unable to restart negotiators: %v", err)) + + // After the start, both sides should still show that they're expecting + // the channel + assert.Equal( + t, sidecar.StateExpectingChannel, testCtx.provider.CurrentState(), + ) + assert.Equal( + t, sidecar.StateExpectingChannel, testCtx.recipient.CurrentState(), + ) + + // We'll now signal to both goroutines that the channel has been + // finalized, at this point, we expect both ticket to transition to the + // terminal state and the goroutines to exit. + go testCtx.confirmSidecarBatch() + + // We expect that both sides now update their state one last time to + // transition the ticket to a completed state, afterwards, they should + // move clean up their mailboxes. + testCtx.assertProviderTicketUpdated(sidecar.StateCompleted) + testCtx.assertProviderMailboxDel() + + testCtx.assertRecipientTicketUpdated(sidecar.StateCompleted) + testCtx.assertRecipientMailboxDel() + + // TODO(roasbeef): send in signals to transition to the final + // terminating state, verify that both sides are shutdown proerly. + + // TODO(roasbeef): on restart starting packet changes +} From fbe1a10d9c3b259258511105f416f65c24101120 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Sun, 23 May 2021 18:48:23 -0700 Subject: [PATCH 12/12] sidecar: add additional test case for message retransmission --- auto_sidecar.go | 15 ++- sidecar_acceptor_test.go | 191 ++++++++++++++++++++++++++++++++++----- 2 files changed, 173 insertions(+), 33 deletions(-) diff --git a/auto_sidecar.go b/auto_sidecar.go index aafd1ee..457e993 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -254,7 +254,7 @@ func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, packetChan := make(chan *sidecar.Ticket, 1) cancelChan := make(chan struct{}) - currentState := startingPkt.CurrentState + atomic.StoreUint32(&a.currentState, uint32(startingPkt.CurrentState)) localTicket := startingPkt.ReceiverTicket // We'll start with a simulated starting message from the sidecar @@ -321,7 +321,7 @@ func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, case newTicket := <-packetChan: newPktState, err := a.stateStepRecipient(ctx, &SidecarPacket{ - CurrentState: currentState, + CurrentState: sidecar.State(a.currentState), ProviderTicket: newTicket, ReceiverTicket: localTicket, }) @@ -470,7 +470,7 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt packetChan := make(chan *sidecar.Ticket, 1) cancelChan := make(chan struct{}) - currentState := startingPkt.CurrentState + atomic.StoreUint32(&a.currentState, uint32(startingPkt.CurrentState)) localTicket := startingPkt.ProviderTicket // We'll start with a simulated starting message from the sidecar @@ -533,10 +533,10 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt // through, so we'll continue until we end up at the // same state (a noop) for { - priorState := currentState + priorState := sidecar.State(atomic.LoadUint32(&a.currentState)) newPktState, err := a.stateStepProvider(ctx, &SidecarPacket{ - CurrentState: currentState, + CurrentState: sidecar.State(a.currentState), ReceiverTicket: newTicket, ProviderTicket: localTicket, }, bid, acct) @@ -545,7 +545,6 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt break } - currentState = newPktState.CurrentState localTicket = newPktState.ProviderTicket atomic.StoreUint32( @@ -553,9 +552,9 @@ func (a *SidecarNegotiator) autoSidecarProvider(ctx context.Context, startingPkt ) switch { - case priorState == currentState: + case priorState == newPktState.CurrentState: fallthrough - case currentState == sidecar.StateExpectingChannel: + case newPktState.CurrentState == sidecar.StateExpectingChannel: break } } diff --git a/sidecar_acceptor_test.go b/sidecar_acceptor_test.go index 2f9f997..fc38ac2 100644 --- a/sidecar_acceptor_test.go +++ b/sidecar_acceptor_test.go @@ -17,6 +17,7 @@ import ( "github.com/lightninglabs/pool/order" "github.com/lightninglabs/pool/sidecar" "github.com/lightningnetwork/lnd/keychain" + "github.com/lightningnetwork/lnd/lntest/wait" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -164,24 +165,28 @@ func TestRegisterSidecar(t *testing.T) { } type mockMailBox struct { - providerChan chan *sidecar.Ticket - providerMsgAck chan struct{} - providerDel chan struct{} + providerChan chan *sidecar.Ticket + providerMsgAck chan struct{} + providerDel chan struct{} + providerDropChan chan struct{} - receiverChan chan *sidecar.Ticket - receiverMsgAck chan struct{} - receiverDel chan struct{} + receiverChan chan *sidecar.Ticket + receiverMsgAck chan struct{} + receiverDel chan struct{} + receiverDropChan chan struct{} } func newMockMailBox() *mockMailBox { return &mockMailBox{ - providerChan: make(chan *sidecar.Ticket), - providerMsgAck: make(chan struct{}), - providerDel: make(chan struct{}), + providerChan: make(chan *sidecar.Ticket), + providerMsgAck: make(chan struct{}), + providerDel: make(chan struct{}), + providerDropChan: make(chan struct{}, 1), - receiverChan: make(chan *sidecar.Ticket), - receiverMsgAck: make(chan struct{}), - receiverDel: make(chan struct{}), + receiverChan: make(chan *sidecar.Ticket), + receiverMsgAck: make(chan struct{}), + receiverDel: make(chan struct{}), + receiverDropChan: make(chan struct{}, 1), } } @@ -190,25 +195,37 @@ func (m *mockMailBox) RecvSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, var ( recvChan chan *sidecar.Ticket + dropChan chan struct{} ackChan chan struct{} ) if provider { recvChan = m.providerChan ackChan = m.providerMsgAck + dropChan = m.providerDropChan } else { recvChan = m.receiverChan ackChan = m.receiverMsgAck + dropChan = m.receiverDropChan } +recvMsg: select { case <-ctx.Done(): return nil, fmt.Errorf("mailbox shutting down") case tkt := <-recvChan: + tktCopy := *tkt - ackChan <- struct{}{} + select { + case ackChan <- struct{}{}: - return tkt, nil + // If we get a signal to drop the message, then we'll just go + // back to receiving as normal. + case <-dropChan: + goto recvMsg + } + + return &tktCopy, nil } } @@ -324,6 +341,9 @@ func (s *sidecarTestCtx) restartAllNegotiators() error { s.provider.quit = make(chan struct{}) s.recipient.quit = make(chan struct{}) + s.provider.cfg.StartingPkt.CurrentState = sidecar.State(s.provider.currentState) + s.recipient.cfg.StartingPkt.CurrentState = sidecar.State(s.recipient.currentState) + if err := s.provider.Start(); err != nil { return err } @@ -331,17 +351,70 @@ func (s *sidecarTestCtx) restartAllNegotiators() error { return s.recipient.Start() } +func (s *sidecarTestCtx) restartProvider() error { + s.provider.Stop() + + s.provider.quit = make(chan struct{}) + + // When we restart the provider in isolation, we'll have their state be + // mapped to the _created_ state (as the SidecarAcceptor would), + // which'll cause them to retransmit their last message. + s.provider.cfg.StartingPkt.CurrentState = sidecar.StateCreated + + return s.provider.Start() +} + +func (s *sidecarTestCtx) restartRecipient() error { + s.recipient.Stop() + + s.recipient.quit = make(chan struct{}) + + s.recipient.cfg.StartingPkt.CurrentState = sidecar.State(s.recipient.currentState) + + return s.recipient.Start() +} + func (s *sidecarTestCtx) assertProviderMsgRecv() { + s.t.Helper() + select { case <-s.mailbox.providerMsgAck: case <-time.After(time.Second * 5): + s.t.Fatalf("no provider msg received") } } func (s *sidecarTestCtx) assertRecipientMsgRecv() { + s.t.Helper() + select { case <-s.mailbox.receiverMsgAck: case <-time.After(time.Second * 5): + s.t.Fatalf("no recipient msg received") + } +} + +func (s *sidecarTestCtx) dropReceiverMessage() { + s.mailbox.receiverDropChan <- struct{}{} +} + +func (s *sidecarTestCtx) dropProviderMessage() { + s.mailbox.providerDropChan <- struct{}{} +} + +func (s *sidecarTestCtx) assertNoProviderMsgsRecvd() { + select { + case <-s.mailbox.providerMsgAck: + s.t.Fatalf("provider should've received no messages") + case <-time.After(time.Second * 1): + } +} + +func (s *sidecarTestCtx) assertNoReceiverMsgsRecvd() { + select { + case <-s.mailbox.receiverMsgAck: + s.t.Fatalf("receiver should've received no messages") + case <-time.After(time.Second * 1): } } @@ -419,6 +492,17 @@ func (s *sidecarTestCtx) assertRecipientMailboxDel() { } } +func (s *sidecarTestCtx) assertNegotiatorStates(providerState, recepientState sidecar.State) { + err := wait.Predicate(func() bool { + return s.provider.CurrentState() == providerState + }, time.Second*5) + assert.NoError(s.t, err) + err = wait.Predicate(func() bool { + return s.recipient.CurrentState() == recepientState + }, time.Second*5) + assert.NoError(s.t, err) +} + func newSidecarTestCtx(t *testing.T) *sidecarTestCtx { mailBox := newMockMailBox() providerDriver := newMockDriver() @@ -531,13 +615,8 @@ func TestAutoSidecarNegotiation(t *testing.T) { // At this point, both sides should be waiting for the channel in its // expected state. - // - // TODO(roasbeef): wrap in wait predicate? - assert.Equal( - t, sidecar.StateExpectingChannel, testCtx.provider.CurrentState(), - ) - assert.Equal( - t, sidecar.StateExpectingChannel, testCtx.recipient.CurrentState(), + testCtx.assertNegotiatorStates( + sidecar.StateExpectingChannel, sidecar.StateExpectingChannel, ) // We'll now simulate a restart on both sides by signalling their @@ -555,6 +634,11 @@ func TestAutoSidecarNegotiation(t *testing.T) { t, sidecar.StateExpectingChannel, testCtx.recipient.CurrentState(), ) + // Finally there should be no additional message sent either since both + // sides should now be in a terminal state + testCtx.assertNoProviderMsgsRecvd() + testCtx.assertNoReceiverMsgsRecvd() + // We'll now signal to both goroutines that the channel has been // finalized, at this point, we expect both ticket to transition to the // terminal state and the goroutines to exit. @@ -569,8 +653,65 @@ func TestAutoSidecarNegotiation(t *testing.T) { testCtx.assertRecipientTicketUpdated(sidecar.StateCompleted) testCtx.assertRecipientMailboxDel() - // TODO(roasbeef): send in signals to transition to the final - // terminating state, verify that both sides are shutdown proerly. - - // TODO(roasbeef): on restart starting packet changes + // Once again, no messages should be received by either side. + testCtx.assertNoProviderMsgsRecvd() + testCtx.assertNoReceiverMsgsRecvd() +} + +// TestAutoSidecarNegotiationRetransmission tests that if either side restarts, +// then the proper message is sent in order to ensure the negotiation state +// machine continues to be progressed. +func TestAutoSidecarNegotiationRetransmission(t *testing.T) { + t.Parallel() + + testCtx := newSidecarTestCtx(t) + + // We'll start our negotiators as usual, however before we start them + // we'll make sure that the message sent by the receiver is never + // received by the provider. + testCtx.dropProviderMessage() + + err := testCtx.startNegotiators() + assert.NoError(t, err, fmt.Errorf("unable to start negotiators: %v", err)) + + // At this point, both sides should still be in their starting state as + // the initial message was never received. + testCtx.assertNegotiatorStates( + sidecar.StateOffered, sidecar.StateRegistered, + ) + + // We'll now restart only the provider. This should cause the provider + // to retransmit a message of their offered ticket, which should cause + // the recipient to re-send their registered ticket. + // + // In order to test the other retransmission case, we'll drop the + // provider's message which carries the ticket in the ordered state. + require.NoError(t, testCtx.restartProvider()) + testCtx.assertRecipientMsgRecv() + + // The provider receive the ticket, then update their local state as + // normal. + testCtx.assertProviderMsgRecv() + testCtx.dropReceiverMessage() + testCtx.assertProviderTicketUpdated(sidecar.StateRegistered) + testCtx.assertBidSubmited() + testCtx.assertProviderTicketUpdated(sidecar.StateExpectingChannel) + + // The provider should now have transitioned to the final state, + // however the recipient should still be in their initial registered + // state as they haven't received any messages yet. + testCtx.assertNegotiatorStates( + sidecar.StateExpectingChannel, sidecar.StateRegistered, + ) + + // We'll now restart the recipient, which should cause them to re-send + // their registered ticket that'll cause the provider to re-send + // _their_ ticket which should conclude the process with the ticket + // being fully finalized. + require.NoError(t, testCtx.restartRecipient()) + testCtx.assertProviderMsgRecv() + + testCtx.assertRecipientMsgRecv() + testCtx.assertRecipientTicketValidated() + testCtx.assertRecipientExpectsChannel() }