🔀 Merge pull request #144 from perun-network/143-settle-implicit-register

Fixes #143 settle implicit register

Author: sebastianst

client/adjudicate.go

@@ -20,6 +20,7 @@ import (
 	"github.com/pkg/errors"
 
 	"perun.network/go-perun/channel"
+	"perun.network/go-perun/log"
 	"perun.network/go-perun/pkg/sync"
 	"perun.network/go-perun/wire"
 )
@@ -75,7 +76,7 @@ func (c *Channel) Watch(h AdjudicatorEventHandler) error {
 
 			// If local version greater than backend version, register local state.
 			if e.Version() < c.State().Version {
-				if err := c.Register(ctx); err != nil {
+				if err := c.registerDispute(ctx); err != nil {
 					return errors.WithMessage(err, "registering")
 				}
 			}
@@ -90,17 +91,17 @@ func (c *Channel) Watch(h AdjudicatorEventHandler) error {
 	return errors.WithMessage(err, "subscription closed")
 }
 
-// Register registers the channel and all its relatives on the adjudicator.
+// registerDispute registers a dispute for the channel and all its relatives.
 //
 // Returns TxTimedoutError when the program times out waiting for a transaction
 // to be mined.
 // Returns ChainNotReachableError if the connection to the blockchain network
 // fails when sending a transaction to / reading from the blockchain.
-func (c *Channel) Register(ctx context.Context) error {
+func (c *Channel) registerDispute(ctx context.Context) error {
 	// If this is not the root, go up one level.
 	// Once we are at the root, we register the whole channel tree together.
 	if c.parent != nil {
-		return c.parent.Register(ctx)
+		return c.parent.registerDispute(ctx)
 	}
 
 	// Lock machines of channel and all subchannels recursively.
@@ -133,13 +134,18 @@ func (c *Channel) Register(ctx context.Context) error {
 	return nil
 }
 
-// ProgressBy progresses the channel state in the adjudicator backend.
+// ForceUpdate enforces a channel update on the adjudicator.
 //
 // Returns TxTimedoutError when the program times out waiting for a transaction
 // to be mined.
 // Returns ChainNotReachableError if the connection to the blockchain network
 // fails when sending a transaction to / reading from the blockchain.
-func (c *Channel) ProgressBy(ctx context.Context, update func(*channel.State)) error {
+func (c *Channel) ForceUpdate(ctx context.Context, update func(*channel.State)) error {
+	err := c.ensureRegistered(ctx)
+	if err != nil {
+		return err
+	}
+
 	// Lock machine
 	if !c.machMtx.TryLockCtx(ctx) {
 		return errors.Errorf("locking machine mutex in time: %v", ctx.Err())
@@ -182,6 +188,13 @@ func (c *Channel) ProgressBy(ctx context.Context, update func(*channel.State)) e
 // Returns ChainNotReachableError if the connection to the blockchain network
 // fails when sending a transaction to / reading from the blockchain.
 func (c *Channel) Settle(ctx context.Context, secondary bool) (err error) {
+	if !c.State().IsFinal {
+		err := c.ensureRegistered(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
 	// Lock machines of channel and all subchannels recursively.
 	l, err := c.tryLockRecursive(ctx)
 	defer l.Unlock()
@@ -396,3 +409,55 @@ func (c *Channel) subChannelStateMap() (states channel.StateMap, err error) {
 	})
 	return
 }
+
+// ensureRegistered ensures that the channel is registered.
+func (c *Channel) ensureRegistered(ctx context.Context) error {
+	phase := c.Phase()
+	if phase == channel.Registered || phase == channel.Progressed {
+		return nil
+	}
+
+	registeredEvents := make(chan *channel.RegisteredEvent)
+
+	// Start event subscription.
+	sub, err := c.adjudicator.Subscribe(ctx, c.Params())
+	if err != nil {
+		return errors.WithMessage(err, "subscribing to adjudicator events")
+	}
+	go waitForRegisteredEvent(sub, registeredEvents)
+
+	// Register.
+	err = c.registerDispute(ctx)
+	if err != nil {
+		return errors.WithMessage(err, "registering")
+	}
+
+	select {
+	case e := <-registeredEvents:
+		err = e.Timeout().Wait(ctx)
+		if err != nil {
+			return err
+		}
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	return nil
+}
+
+// waitForRegisteredEvent waits until a RegisteredEvent has been observed.
+func waitForRegisteredEvent(sub channel.AdjudicatorSubscription, events chan<- *channel.RegisteredEvent) {
+	defer func() {
+		err := sub.Close()
+		if err != nil {
+			log.Warn("Subscription closed with error:", err)
+		}
+	}()
+
+	// Scan for registered event.
+	for e := sub.Next(); e != nil; e = sub.Next() {
+		if e, ok := e.(*channel.RegisteredEvent); ok {
+			events <- e
+			return
+		}
+	}
+}

client/test/channel.go

@@ -184,10 +184,6 @@ func (ch *paymentChannel) settleImpl(secondary bool) {
 	ctx, cancel := context.WithTimeout(context.Background(), ch.r.timeout)
 	defer cancel()
 
-	if ch.IsLedgerChannel() || !ch.State().IsFinal {
-		assert.NoError(ch.Register(ctx))
-	}
-
 	assert.NoError(ch.Settle(ctx, secondary))
 	ch.assertBals(ch.State())
 

client/test/progression.go

@@ -91,14 +91,8 @@ func (r *Paul) exec(_cfg ExecConfig, ch *paymentChannel) {
 
 	r.waitStage() // wait for setup complete
 
-	// register
-	assert.NoError(ch.Register(ctx), "registering")
-	regEvent := <-r.registered
-
-	assert.NoError(regEvent.Timeout().Wait(ctx), "waiting for refutation timeout")
-
 	// progress
-	assert.NoError(ch.ProgressBy(ctx, func(s *channel.State) {
+	assert.NoError(ch.ForceUpdate(ctx, func(s *channel.State) {
 		bal := func(user channel.Index) int64 {
 			return s.Balances[assetIdx][user].Int64()
 		}
@@ -168,7 +162,7 @@ func (r *Paula) exec(_cfg ExecConfig, ch *paymentChannel, _ *acceptNextPropHandl
 	r.t.Logf("%v received progression confirmation 1", r.setup.Name)
 
 	// we progress
-	assert.NoError(ch.ProgressBy(ctx, func(s *channel.State) {
+	assert.NoError(ch.ForceUpdate(ctx, func(s *channel.State) {
 		bal := func(user channel.Index) int64 {
 			return s.Balances[assetIdx][user].Int64()
 		}

client/testchannel.go

@@ -14,7 +14,11 @@
 
 package client
 
-import "perun.network/go-perun/channel"
+import (
+	"context"
+
+	"perun.network/go-perun/channel"
+)
 
 // TestChannel grants access to the `AdjudicatorRequest` which is otherwise
 // hidden by `Channel`. Behaves like a `Channel` in all other cases.
@@ -33,3 +37,8 @@ func NewTestChannel(c *Channel) *TestChannel {
 func (c *TestChannel) AdjudicatorReq() channel.AdjudicatorReq {
 	return c.machine.AdjudicatorReq()
 }
+
+// Register exposes dispute registration.
+func (c *TestChannel) Register(ctx context.Context) error {
+	return c.Channel.registerDispute(ctx)
+}

client/virtual_channel_test.go

@@ -80,7 +80,7 @@ func TestVirtualChannelsDispute(t *testing.T) {
 	chs := []*client.Channel{vct.chAliceIngrid, vct.chIngridAlice, vct.chBobIngrid, vct.chIngridBob}
 	// Register the channels in a random order.
 	for _, i := range rand.Perm(len(chs)) {
-		err := chs[i].Register(ctx)
+		err := client.NewTestChannel(chs[i]).Register(ctx)
 		assert.NoErrorf(err, "register channel: %d", i)
 	}