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 } diff --git a/auto_sidecar.go b/auto_sidecar.go index c79652c..457e993 100644 --- a/auto_sidecar.go +++ b/auto_sidecar.go @@ -1,20 +1,59 @@ package pool import ( - "bytes" "context" + "crypto/sha512" "errors" "fmt" + "sync" + "sync/atomic" "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" ) +// 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. +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 + + // 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. // Note that the state of the negotiator, and the ticket may differ, this is // what will trigger a state transition. @@ -34,7 +73,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 @@ -53,21 +91,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,88 +118,143 @@ 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 -// 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 +} + +// 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 { + currentState uint32 + + cfg AutoAcceptorConfig + + wg sync.WaitGroup + + ticketFinalized chan struct{} + quit chan struct{} + + stopOnce sync.Once +} + +// NewSidecarNegotiator returns a new instance of the sidecar negotiator given +// a valid config. +func NewSidecarNegotiator(cfg AutoAcceptorConfig) *SidecarNegotiator { + return &SidecarNegotiator{ + cfg: cfg, + currentState: uint32(cfg.StartingPkt.CurrentState), + ticketFinalized: make(chan struct{}), + quit: make(chan struct{}), } +} - 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.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)) + return nil } -// 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 +// 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: } - return statusCode.Code() == codes.AlreadyExists + 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 *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { +func (a *SidecarNegotiator) autoSidecarReceiver(ctx context.Context, + startingPkt *SidecarPacket) { + defer a.wg.Done() 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 @@ -167,17 +263,20 @@ 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( - 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) @@ -196,10 +295,11 @@ 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? log.Error(err) return } @@ -220,8 +320,8 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { select { case newTicket := <-packetChan: - newPktState, err := a.stateStepRecipient(&SidecarPacket{ - CurrentState: currentState, + newPktState, err := a.stateStepRecipient(ctx, &SidecarPacket{ + CurrentState: sidecar.State(a.currentState), ProviderTicket: newTicket, ReceiverTicket: localTicket, }) @@ -230,17 +330,34 @@ func (a *SidecarAcceptor) autoSidecarReceiver(startingPkt *SidecarPacket) { 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, - // 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 + } + + // 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 } @@ -254,8 +371,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 { @@ -280,7 +397,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) } @@ -303,10 +420,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) @@ -318,14 +432,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, @@ -337,7 +451,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) } } @@ -345,7 +459,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() @@ -356,12 +470,15 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, 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 - // 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. @@ -375,9 +492,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 @@ -391,8 +506,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) @@ -418,12 +533,10 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, // 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)) - log.Infof("step=%v", currentState) - - newPktState, err := a.stateStepProvider(&SidecarPacket{ - CurrentState: currentState, + newPktState, err := a.stateStepProvider(ctx, &SidecarPacket{ + CurrentState: sidecar.State(a.currentState), ReceiverTicket: newTicket, ProviderTicket: localTicket, }, bid, acct) @@ -432,40 +545,60 @@ func (a *SidecarAcceptor) autoSidecarProvider(startingPkt *SidecarPacket, break } - currentState = newPktState.CurrentState localTicket = newPktState.ProviderTicket + atomic.StoreUint32( + &a.currentState, uint32(newPktState.CurrentState), + ) + switch { - case priorState == currentState: + case priorState == newPktState.CurrentState: fallthrough - case currentState == sidecar.StateExpectingChannel: + case newPktState.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.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 } } } +// 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. -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 @@ -478,7 +611,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 } @@ -502,7 +635,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) @@ -526,8 +659,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 @@ -571,7 +704,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) @@ -584,7 +717,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/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) diff --git a/sidecar_acceptor.go b/sidecar_acceptor.go index c1ef9a6..9e75b80 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 @@ -45,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 @@ -84,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), } } @@ -132,11 +141,29 @@ 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, }) + if err := autoAcceptor.Start(); err != nil { + 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. @@ -162,18 +189,39 @@ 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, + }) + if err := autoAcceptor.Start(); err != nil { + 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) } + 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: @@ -238,6 +286,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) @@ -374,30 +426,42 @@ 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, }) - return nil + 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() } -// 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) @@ -422,14 +486,30 @@ 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) + autoAcceptor := NewSidecarNegotiator(AutoAcceptorConfig{ + Provider: true, + ProviderBid: bid, + StartingPkt: &SidecarPacket{ + CurrentState: sidecar.StateOffered, + ProviderTicket: ticket, + ReceiverTicket: ticket, + }, + ProviderAccount: acct, + Driver: a, + MailBox: a, + }) - return nil + 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() } // handleServerMessage reacts to a message sent by the server and sends back the @@ -620,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) } } @@ -686,3 +782,134 @@ 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)) +} + +// 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. +func isErrAlreadyExists(err error) bool { + statusCode, ok := status.FromError(err) + if !ok { + return false + } + + return statusCode.Code() == codes.AlreadyExists +} diff --git a/sidecar_acceptor_test.go b/sidecar_acceptor_test.go index e933d50..fc38ac2 100644 --- a/sidecar_acceptor_test.go +++ b/sidecar_acceptor_test.go @@ -2,16 +2,23 @@ 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/lightningnetwork/lnd/lntest/wait" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -156,3 +163,555 @@ func TestRegisterSidecar(t *testing.T) { cleanup() } } + +type mockMailBox struct { + providerChan chan *sidecar.Ticket + providerMsgAck chan struct{} + providerDel chan struct{} + providerDropChan 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{}), + providerDropChan: make(chan struct{}, 1), + + receiverChan: make(chan *sidecar.Ticket), + receiverMsgAck: make(chan struct{}), + receiverDel: make(chan struct{}), + receiverDropChan: make(chan struct{}, 1), + } +} + +func (m *mockMailBox) RecvSidecarPkt(ctx context.Context, pkt *sidecar.Ticket, + provider bool) (*sidecar.Ticket, error) { + + 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 + + select { + case ackChan <- struct{}{}: + + // 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 + } +} + +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{}) + + 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 + } + + 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): + } +} + +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 (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() + 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. + testCtx.assertNegotiatorStates( + sidecar.StateExpectingChannel, sidecar.StateExpectingChannel, + ) + + // 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(), + ) + + // 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. + 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() + + // 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() +}