🐛 Cache first channel update

Author: matthiasgeihs

client/client.go

@@ -34,14 +34,15 @@ import (
 //
 // Currently, only the two-party protocol is fully implemented.
 type Client struct {
-	address     wire.Address
-	conn        clientConn
-	channels    chanRegistry
-	funder      channel.Funder
-	adjudicator channel.Adjudicator
-	wallet      wallet.Wallet
-	pr          persistence.PersistRestorer
-	log         log.Logger // structured logger for this client
+	address       wire.Address
+	conn          clientConn
+	channels      chanRegistry
+	funder        channel.Funder
+	adjudicator   channel.Adjudicator
+	wallet        wallet.Wallet
+	pr            persistence.PersistRestorer
+	log           log.Logger // structured logger for this client
+	version1Cache version1Cache
 
 	sync.Closer
 }

client/proposal.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"sync"
 
 	"github.com/pkg/errors"
 
@@ -140,13 +141,16 @@ func (c *Client) ProposeChannel(ctx context.Context, prop ChannelProposal) (*Cha
 	}
 
 	// 2. send proposal, wait for response, create channel object
+	c.enableVer1Cache() // cache version 1 updates until channel is opened
 	ch, err := c.proposeTwoPartyChannel(ctx, prop)
 	if err != nil {
 		return nil, errors.WithMessage(err, "channel proposal")
 	}
 
 	// 3. fund
-	return ch, c.fundChannel(ctx, ch, prop)
+	fundingErr := c.fundChannel(ctx, ch, prop)
+	c.releaseVer1Cache() // replay cached version 1 updates
+	return ch, fundingErr
 }
 
 // handleChannelProposal implements the receiving side of the (currently)
@@ -175,11 +179,15 @@ func (c *Client) handleChannelProposalAcc(
 		return ch, errors.WithMessage(err, "validating channel proposal acceptance")
 	}
 
+	c.enableVer1Cache() // cache version 1 updates
+
 	if ch, err = c.acceptChannelProposal(ctx, prop, p, acc); err != nil {
 		return ch, errors.WithMessage(err, "accept channel proposal")
 	}
 
-	return ch, c.fundChannel(ctx, ch, prop)
+	fundingErr := c.fundChannel(ctx, ch, prop)
+	c.releaseVer1Cache() // replay cached version 1 updates
+	return ch, fundingErr
 }
 
 func (c *Client) acceptChannelProposal(
@@ -549,6 +557,40 @@ func enableVer0Cache(ctx context.Context, c wire.Cacher) {
 	})
 }
 
+func (c *Client) enableVer1Cache() {
+	c.log.Trace("Enabling version 1 cache")
+
+	c.version1Cache.mu.Lock()
+	defer c.version1Cache.mu.Unlock()
+
+	c.version1Cache.enabled++
+}
+
+func (c *Client) releaseVer1Cache() {
+	c.log.Trace("Releasing version 1 cache")
+
+	c.version1Cache.mu.Lock()
+	defer c.version1Cache.mu.Unlock()
+
+	c.version1Cache.enabled--
+	for _, u := range c.version1Cache.cache {
+		go c.handleChannelUpdate(u.uh, u.p, u.m)
+	}
+	c.version1Cache.cache = nil
+}
+
+type version1Cache struct {
+	mu      sync.Mutex
+	enabled uint // counter to support concurrent channel openings
+	cache   []cachedUpdate
+}
+
+type cachedUpdate struct {
+	uh UpdateHandler
+	p  wire.Address
+	m  *msgChannelUpdate
+}
+
 // Error implements the error interface.
 func (e PeerRejectedProposalError) Error() string {
 	return fmt.Sprintf("channel proposal rejected: %s", e.reason)

client/update.go

@@ -34,13 +34,31 @@ import (
 func (c *Client) handleChannelUpdate(uh UpdateHandler, p wire.Address, m *msgChannelUpdate) {
 	ch, ok := c.channels.Get(m.ID())
 	if !ok {
-		c.logChan(m.ID()).WithField("peer", p).Error("received update for unknown channel")
+		if !c.cacheVersion1Update(uh, p, m) {
+			c.logChan(m.ID()).WithField("peer", p).Error("received update for unknown channel")
+		}
 		return
 	}
 	pidx := ch.Idx() ^ 1
 	ch.handleUpdateReq(pidx, m, uh)
 }
 
+func (c *Client) cacheVersion1Update(uh UpdateHandler, p wire.Address, m *msgChannelUpdate) bool {
+	c.version1Cache.mu.Lock()
+	defer c.version1Cache.mu.Unlock()
+
+	if !(m.State.Version == 1 && c.version1Cache.enabled > 0) {
+		return false
+	}
+
+	c.version1Cache.cache = append(c.version1Cache.cache, cachedUpdate{
+		uh: uh,
+		p:  p,
+		m:  m,
+	})
+	return true
+}
+
 type (
 	// ChannelUpdate is a channel update proposal.
 	ChannelUpdate struct {