Merge pull request #4 from perun-network/cache-first-update

🔀 Cache first channel update

Author: ggwpez

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/client_role_test.go

@@ -38,11 +38,21 @@ func NewSetups(rng *rand.Rand, names []string) []ctest.RoleSetup {
 
 	for i := 0; i < n; i++ {
 		acc := wtest.NewRandomAccount(rng)
+
+		// The use of a delayed funder simulates that channel participants may
+		// receive their funding confirmation at different times.
+		var funder channel.Funder
+		if i == 0 {
+			funder = &logFunderWithDelay{log.WithField("role", names[i])}
+		} else {
+			funder = &logFunder{log.WithField("role", names[i])}
+		}
+
 		setup[i] = ctest.RoleSetup{
 			Name:        names[i],
 			Identity:    acc,
 			Bus:         bus,
-			Funder:      &logFunder{log.WithField("role", names[i])},
+			Funder:      funder,
 			Adjudicator: &logAdjudicator{log.WithField("role", names[i]), sync.RWMutex{}, nil},
 			Wallet:      wtest.NewWallet(),
 			Timeout:     roleOperationTimeout,
@@ -57,6 +67,10 @@ type (
 		log log.Logger
 	}
 
+	logFunderWithDelay struct {
+		log log.Logger
+	}
+
 	logAdjudicator struct {
 		log         log.Logger
 		mu          sync.RWMutex
@@ -69,6 +83,12 @@ func (f *logFunder) Fund(_ context.Context, req channel.FundingReq) error {
 	return nil
 }
 
+func (f *logFunderWithDelay) Fund(_ context.Context, req channel.FundingReq) error {
+	time.Sleep(100 * time.Millisecond)
+	f.log.Infof("Funding: %v", req)
+	return nil
+}
+
 func (a *logAdjudicator) Register(_ context.Context, req channel.AdjudicatorReq) error {
 	a.log.Infof("Register: %v", req)
 	e := channel.NewRegisteredEvent(

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/test/alice.go

@@ -35,7 +35,7 @@ type Alice struct {
 
 // NewAlice creates a new Proposer that executes the Alice protocol.
 func NewAlice(setup RoleSetup, t *testing.T) *Alice {
-	return &Alice{Proposer: *NewProposer(setup, t, 4)}
+	return &Alice{Proposer: *NewProposer(setup, t, 3)}
 }
 
 // Execute executes the Alice protocol.
@@ -46,8 +46,6 @@ func (r *Alice) Execute(cfg ExecConfig) {
 func (r *Alice) exec(_cfg ExecConfig, ch *paymentChannel) {
 	cfg := _cfg.(*AliceBobExecConfig)
 	we, them := r.Idxs(cfg.Peers())
-	// 1st stage - channel controller set up
-	r.waitStage()
 
 	// 1st Alice receives some updates from Bob
 	for i := 0; i < cfg.NumPayments[them]; i++ {

client/test/bob.go

@@ -26,7 +26,7 @@ type Bob struct {
 
 // NewBob creates a new Responder that executes the Bob protocol.
 func NewBob(setup RoleSetup, t *testing.T) *Bob {
-	return &Bob{Responder: *NewResponder(setup, t, 4)}
+	return &Bob{Responder: *NewResponder(setup, t, 3)}
 }
 
 // Execute executes the Bob protocol.
@@ -38,9 +38,6 @@ func (r *Bob) exec(_cfg ExecConfig, ch *paymentChannel, propHandler *acceptNextP
 	cfg := _cfg.(*AliceBobExecConfig)
 	we, them := r.Idxs(cfg.Peers())
 
-	// 1st stage - channel controller set up
-	r.waitStage()
-
 	// 1st Bob sends some updates to Alice
 	for i := 0; i < cfg.NumPayments[we]; i++ {
 		ch.sendTransfer(cfg.TxAmounts[we], fmt.Sprintf("Bob#%d", i))

client/test/proposer.go

@@ -36,6 +36,10 @@ func (r *Proposer) Execute(cfg ExecConfig, exec func(ExecConfig, *paymentChannel
 	rng := pkgtest.Prng(r.t, "proposer")
 	assert := assert.New(r.t)
 
+	// ignore proposal handler since Proposer doesn't accept any incoming channels
+	_, wait := r.GoHandle(rng)
+	defer wait()
+
 	prop := r.LedgerChannelProposal(rng, cfg)
 	ch, err := r.ProposeChannel(prop)
 	assert.NoError(err)
@@ -45,10 +49,6 @@ func (r *Proposer) Execute(cfg ExecConfig, exec func(ExecConfig, *paymentChannel
 	}
 	r.log.Infof("New Channel opened: %v", ch.Channel)
 
-	// ignore proposal handler since Proposer doesn't accept any incoming channels
-	_, wait := r.GoHandle(rng)
-	defer wait()
-
 	exec(cfg, ch)
 
 	ch.Close() // May or may not already be closed due to channelConn closing.

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 {