Merge pull request #337 from perun-network/multi-ledger

Multi ledger

Author: matthiasgeihs

.golangci.yml

@@ -24,6 +24,8 @@ linters:
     - wsl # Formatting disabled for now.
     - gochecknoinits # We use init functions.
     - gci # We have our own import order.
+    - goerr113 # We do not strictly require static errors.
+    - promlinter # Disabled because unstable.
 
     # These could be enabled in the future:
     - ifshort # we often don't use `if err := …` for readability.

channel/multi/adjudicator.go

@@ -0,0 +1,114 @@
+// Copyright 2022 - See NOTICE file for copyright holders.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package multi
+
+import (
+	"context"
+	"fmt"
+
+	"perun.network/go-perun/channel"
+)
+
+// Adjudicator is a multi-ledger adjudicator.
+type Adjudicator struct {
+	adjudicators map[LedgerIDMapKey]channel.Adjudicator
+}
+
+// NewAdjudicator creates a new adjudicator.
+func NewAdjudicator() *Adjudicator {
+	return &Adjudicator{
+		adjudicators: make(map[LedgerIDMapKey]channel.Adjudicator),
+	}
+}
+
+// RegisterAdjudicator registers an adjudicator for a given ledger.
+func (a *Adjudicator) RegisterAdjudicator(l LedgerID, la channel.Adjudicator) {
+	a.adjudicators[l.MapKey()] = la
+}
+
+// Register registers a multi-ledger channel. It dispatches Register calls to
+// all relevant adjudicators. If any of the calls fails, the method returns an
+// error.
+func (a *Adjudicator) Register(ctx context.Context, req channel.AdjudicatorReq, subStates []channel.SignedState) error {
+	ledgers, err := assets(req.Tx.Assets).LedgerIDs()
+	if err != nil {
+		return err
+	}
+
+	err = a.dispatch(ledgers, func(la channel.Adjudicator) error {
+		return la.Register(ctx, req, subStates)
+	})
+	return err
+}
+
+// Progress progresses the state of a multi-ledger channel. It dispatches
+// Progress calls to all relevant adjudicators. If any of the calls fails, the
+// method returns an error.
+func (a *Adjudicator) Progress(ctx context.Context, req channel.ProgressReq) error {
+	ledgers, err := assets(req.Tx.Assets).LedgerIDs()
+	if err != nil {
+		return err
+	}
+
+	err = a.dispatch(ledgers, func(la channel.Adjudicator) error {
+		return la.Progress(ctx, req)
+	})
+	return err
+}
+
+// Withdraw withdraws the funds from a multi-ledger channel. It dispatches
+// Withdraw calls to all relevant adjudicators. If any of the calls fails, the
+// method returns an error.
+func (a *Adjudicator) Withdraw(ctx context.Context, req channel.AdjudicatorReq, subStates channel.StateMap) error {
+	ledgers, err := assets(req.Tx.Assets).LedgerIDs()
+	if err != nil {
+		return err
+	}
+
+	err = a.dispatch(ledgers, func(la channel.Adjudicator) error {
+		return la.Withdraw(ctx, req, subStates)
+	})
+	return err
+}
+
+// dispatch dispatches an adjudicator call on all given ledgers.
+func (a *Adjudicator) dispatch(ledgers []LedgerID, f func(channel.Adjudicator) error) error {
+	n := len(ledgers)
+	errs := make(chan error, n)
+	for _, l := range ledgers {
+		go func(l LedgerID) {
+			err := func() error {
+				id := l.MapKey()
+				la, ok := a.adjudicators[id]
+				if !ok {
+					return fmt.Errorf("Adjudicator not found for ledger %v", id)
+				}
+
+				err := f(la)
+				return err
+			}()
+			errs <- err
+		}(l)
+	}
+
+	for i := 0; i < n; i++ {
+		err := <-errs
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

channel/multi/asset.go

@@ -0,0 +1,81 @@
+// Copyright 2022 - See NOTICE file for copyright holders.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package multi
+
+import (
+	"fmt"
+
+	"perun.network/go-perun/channel"
+)
+
+type (
+	// Asset defines a multi-ledger asset.
+	Asset interface {
+		channel.Asset
+		LedgerID() LedgerID
+	}
+
+	// LedgerIDMapKey is the map key representation of a ledger identifier.
+	LedgerIDMapKey string
+
+	// LedgerID represents a ledger identifier.
+	LedgerID interface {
+		MapKey() LedgerIDMapKey
+	}
+
+	assets []channel.Asset
+)
+
+// LedgerIDs returns the identifiers of the associated ledgers.
+func (a assets) LedgerIDs() ([]LedgerID, error) {
+	ids := make(map[LedgerIDMapKey]LedgerID)
+	for _, asset := range a {
+		ma, ok := asset.(Asset)
+		if !ok {
+			return nil, fmt.Errorf("wrong asset type: expected multi.Asset, got %T", a)
+		}
+
+		id := ma.LedgerID()
+		ids[id.MapKey()] = id
+	}
+
+	idsArray := make([]LedgerID, len(ids))
+	i := 0
+	for _, v := range ids {
+		idsArray[i] = v
+		i++
+	}
+
+	return idsArray, nil
+}
+
+// IsMultiLedgerAssets returns whether the assets are from multiple ledgers.
+func IsMultiLedgerAssets(assets []channel.Asset) bool {
+	hasMulti := false
+	var id LedgerID
+	for _, asset := range assets {
+		multiAsset, ok := asset.(Asset)
+		switch {
+		case !ok:
+			continue
+		case !hasMulti:
+			hasMulti = true
+			id = multiAsset.LedgerID()
+		case id.MapKey() != multiAsset.LedgerID().MapKey():
+			return true
+		}
+	}
+	return false
+}

channel/multi/doc.go

@@ -0,0 +1,16 @@
+// Copyright 2022 - See NOTICE file for copyright holders.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package multi provides a multi-ledger funder and adjudicator.
+package multi // import "perun.network/go-perun/channel/multi"

channel/multi/funder.go

@@ -0,0 +1,81 @@
+// Copyright 2022 - See NOTICE file for copyright holders.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package multi
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"perun.network/go-perun/channel"
+)
+
+// Funder is a multi-ledger funder.
+type Funder struct {
+	funders map[LedgerIDMapKey]channel.Funder
+}
+
+// NewFunder creates a new funder.
+func NewFunder() *Funder {
+	return &Funder{
+		funders: make(map[LedgerIDMapKey]channel.Funder),
+	}
+}
+
+// RegisterFunder registers a funder for a given ledger.
+func (f *Funder) RegisterFunder(l LedgerID, lf channel.Funder) {
+	f.funders[l.MapKey()] = lf
+}
+
+// Fund funds a multi-ledger channel. It dispatches funding calls to all
+// relevant registered funders. It waits until all participants have funded the
+// channel. If any of the funder calls fails, the method returns an error.
+func (f *Funder) Fund(ctx context.Context, request channel.FundingReq) error {
+	// Define funding timeout.
+	d := time.Duration(request.Params.ChallengeDuration) * time.Second
+	ctx, cancel := context.WithTimeout(ctx, d)
+	defer cancel()
+
+	ledgers, err := assets(request.State.Assets).LedgerIDs()
+	if err != nil {
+		return err
+	}
+
+	n := len(ledgers)
+	errs := make(chan error, n)
+	for _, l := range ledgers {
+		go func(l LedgerID) {
+			errs <- func() error {
+				id := l.MapKey()
+				lf, ok := f.funders[id]
+				if !ok {
+					return fmt.Errorf("Funder not found for ledger %v", id)
+				}
+
+				err := lf.Fund(ctx, request)
+				return err
+			}()
+		}(l)
+	}
+
+	for i := 0; i < n; i++ {
+		err := <-errs
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

channel/multi/subscription.go

@@ -0,0 +1,80 @@
+// Copyright 2022 - See NOTICE file for copyright holders.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package multi
+
+import (
+	"context"
+
+	"perun.network/go-perun/channel"
+)
+
+// Subscribe creates a new multi-ledger AdjudicatorSubscription.
+func (a *Adjudicator) Subscribe(ctx context.Context, chID channel.ID) (channel.AdjudicatorSubscription, error) {
+	asub := &AdjudicatorSubscription{
+		events: make(chan channel.AdjudicatorEvent),
+		errors: make(chan error),
+		subs:   []channel.AdjudicatorSubscription{},
+	}
+
+	for _, la := range a.adjudicators {
+		sub, err := la.Subscribe(ctx, chID)
+		if err != nil {
+			asub.Close()
+			return nil, err
+		}
+		asub.subs = append(asub.subs, sub)
+
+		go func() {
+			asub.events <- sub.Next()
+		}()
+
+		go func() {
+			asub.errors <- sub.Err()
+		}()
+	}
+
+	return asub, nil
+}
+
+// AdjudicatorSubscription is a multi-ledger adjudicator subscription.
+type AdjudicatorSubscription struct {
+	subs   []channel.AdjudicatorSubscription
+	events chan channel.AdjudicatorEvent
+	errors chan error
+}
+
+// Next returns the next event.
+func (s *AdjudicatorSubscription) Next() channel.AdjudicatorEvent {
+	return <-s.events
+}
+
+// Err blocks until an error occurred and returns it.
+func (s *AdjudicatorSubscription) Err() error {
+	for i := 0; i < len(s.subs); i++ {
+		err := <-s.errors
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Close closes the subscription.
+func (s *AdjudicatorSubscription) Close() error {
+	for _, sub := range s.subs {
+		sub.Close()
+	}
+	return nil
+}

channel/persistence/test/channel.go

@@ -54,7 +54,8 @@ func NewRandomChannel(
 	user channel.Index,
 	peers []wire.Address,
 	parent *Channel,
-	rng *rand.Rand) (c *Channel) {
+	rng *rand.Rand,
+) (c *Channel) {
 	accs, parts := wtest.NewRandomAccounts(rng, len(peers))
 	params := ctest.NewRandomParams(rng, ctest.WithParts(parts...))
 	csm, err := channel.NewStateMachine(accs[0], *params)