Merge pull request #258 from lightninglabs/sidecar-follow-ups

multi: sidecar bug fixes and additional tests

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
 }
 

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)

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)

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
+}

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()
+}