Merge pull request #36 from perun-network/generic-events-sub

Generic events sub

Author: matthiasgeihs

backend/ethereum/channel/funder.go

@@ -27,7 +27,7 @@ import (
 	"github.com/pkg/errors"
 
 	"perun.network/go-perun/backend/ethereum/bindings/assetholder"
-	cherrors "perun.network/go-perun/backend/ethereum/channel/errors"
+	"perun.network/go-perun/backend/ethereum/subscription"
 	"perun.network/go-perun/backend/ethereum/wallet"
 	"perun.network/go-perun/channel"
 	"perun.network/go-perun/client"
@@ -184,7 +184,7 @@ func (f *Funder) sendFundingTx(ctx context.Context, request channel.FundingReq,
 	// nolint: gocritic
 	if bal == nil || bal.Sign() <= 0 {
 		f.log.WithFields(log.Fields{"channel": request.Params.ID(), "idx": request.Idx}).Debug("Skipped zero funding.")
-	} else if alreadyFunded, err := checkFunded(ctx, bal, contract, fundingID); err != nil {
+	} else if alreadyFunded, err := f.checkFunded(ctx, bal, contract, fundingID); err != nil {
 		return nil, errors.WithMessage(err, "checking funded")
 	} else if alreadyFunded {
 		f.log.WithFields(log.Fields{"channel": request.Params.ID(), "idx": request.Idx}).Debug("Skipped second funding.")
@@ -210,76 +210,61 @@ func (f *Funder) deposit(ctx context.Context, bal *big.Int, asset Asset, funding
 }
 
 // checkFunded returns whether `fundingID` holds at least `amount` funds.
-func checkFunded(ctx context.Context, amount *big.Int, asset assetHolder, fundingID [32]byte) (bool, error) {
-	iter, err := filterFunds(ctx, asset, fundingID)
+func (f *Funder) checkFunded(ctx context.Context, amount *big.Int, asset assetHolder, fundingID [32]byte) (bool, error) {
+	deposited := make(chan *subscription.Event, 10)
+	subErr := make(chan error, 1)
+	// Subscribe to events.
+	sub, err := f.depositedSub(ctx, asset.contract, fundingID)
 	if err != nil {
-		return false, errors.WithMessagef(err, "filtering old Funding events for asset %d", asset.assetIndex)
+		return false, errors.WithMessage(err, "subscribing to deposited event")
 	}
-	// nolint:errcheck
-	defer iter.Close()
+	defer sub.Close()
+	// Read from the sub.
+	go func() {
+		defer close(deposited)
+		subErr <- sub.ReadPast(ctx, deposited)
+	}()
 
 	left := new(big.Int).Set(amount)
-	for iter.Next() {
-		left.Sub(left, iter.Event.Amount)
+	for _event := range deposited {
+		event := _event.Data.(*assetholder.AssetHolderDeposited)
+		left.Sub(left, event.Amount)
 	}
-	return left.Sign() != 1, errors.Wrap(iter.Error(), "iterator")
+	return left.Sign() != 1, errors.WithMessagef(<-subErr, "filtering old Funding events for asset %d", asset.assetIndex)
 }
 
-func filterFunds(ctx context.Context, asset assetHolder, fundingIDs ...[32]byte) (*assetholder.AssetHolderDepositedIterator, error) {
-	// Filter
-	filterOpts := bind.FilterOpts{
-		Start:   uint64(1),
-		End:     nil,
-		Context: ctx}
-	iter, err := asset.FilterDeposited(&filterOpts, fundingIDs)
-	if err != nil {
-		err = cherrors.CheckIsChainNotReachableError(err)
-		return nil, errors.WithMessage(err, "filtering deposited events")
+func (f *Funder) depositedSub(ctx context.Context, contract *bind.BoundContract, fundingIDs ...[32]byte) (*subscription.EventSub, error) {
+	filter := make([]interface{}, len(fundingIDs))
+	for i, fundingID := range fundingIDs {
+		filter[i] = fundingID
 	}
-
-	return iter, nil
+	event := func() *subscription.Event {
+		return &subscription.Event{
+			Name:   "Deposited",
+			Data:   new(assetholder.AssetHolderDeposited),
+			Filter: [][]interface{}{filter},
+		}
+	}
+	sub, err := subscription.NewEventSub(ctx, f, contract, event, startBlockOffset)
+	return sub, errors.WithMessage(err, "subscribing to deposited event")
 }
 
 // waitForFundingConfirmation waits for the confirmation events on the blockchain that
 // both we and all peers successfully funded the channel for the specified asset
 // according to the funding agreement.
 // nolint: funlen
 func (f *Funder) waitForFundingConfirmation(ctx context.Context, request channel.FundingReq, asset assetHolder, fundingIDs [][32]byte) error {
-	deposited := make(chan *assetholder.AssetHolderDeposited)
-	// Watch new events
-	watchOpts, err := f.NewWatchOpts(ctx)
-	if err != nil {
-		return errors.WithMessage(err, "error creating watchopts")
-	}
-	sub, err := asset.WatchDeposited(watchOpts, deposited, fundingIDs)
+	deposited := make(chan *subscription.Event)
+	subErr := make(chan error, 1)
+	// Subscribe to events.
+	sub, err := f.depositedSub(ctx, asset.contract, fundingIDs...)
 	if err != nil {
-		err = cherrors.CheckIsChainNotReachableError(err)
-		return errors.WithMessagef(err, "WatchDeposit on asset %d failed", asset.assetIndex)
+		return errors.WithMessage(err, "subscribing to deposited event")
 	}
-	defer sub.Unsubscribe()
-
-	// we let the filter queries and all subscription errors write into this error
-	// channel.
-	errChan := make(chan error, 1)
+	defer sub.Close()
+	// Read from the sub.
 	go func() {
-		err := <-sub.Err()
-		if err != nil {
-			err = cherrors.CheckIsChainNotReachableError(err)
-		}
-		errChan <- errors.WithMessagef(err, "subscription for asset %d", asset.assetIndex)
-	}()
-
-	// Query all old funding events
-	go func() {
-		iter, err := filterFunds(ctx, asset, fundingIDs...)
-		if err != nil {
-			errChan <- errors.WithMessagef(err, "filtering old Deposited events for asset %d", asset.assetIndex)
-			return
-		}
-		defer iter.Close() // nolint: errcheck
-		for iter.Next() {
-			deposited <- iter.Event
-		}
+		subErr <- sub.ReadAll(ctx, deposited)
 	}()
 
 	// The allocation that all participants agreed on.
@@ -290,7 +275,8 @@ func (f *Funder) waitForFundingConfirmation(ctx context.Context, request channel
 	// Wait for all non-zero funding requests
 	for N > 0 {
 		select {
-		case event := <-deposited:
+		case rawEvent := <-deposited:
+			event := rawEvent.Data.(*assetholder.AssetHolderDeposited)
 			log := f.log.WithField("fundingID", event.FundingID)
 
 			// Calculate the position in the participant array.
@@ -320,7 +306,7 @@ func (f *Funder) waitForFundingConfirmation(ctx context.Context, request channel
 				return &channel.AssetFundingError{Asset: asset.assetIndex, TimedOutPeers: indices}
 			}
 			return nil
-		case err := <-errChan:
+		case err := <-subErr:
 			return err
 		}
 	}

backend/ethereum/subscription/doc.go

@@ -0,0 +1,16 @@
+// Copyright 2021 - 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 subscription contains generic event subscriptions.
+package subscription // import "perun.network/go-perun/backend/ethereum/subscription"

backend/ethereum/subscription/eventsub.go

@@ -0,0 +1,226 @@
+// Copyright 2021 - 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 subscription
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/event"
+	cherrors "perun.network/go-perun/backend/ethereum/channel/errors"
+)
+
+type (
+	// EventSub generic event subscription.
+	// Can be used on any Contract with any Event.
+	// This EventSub does not prevent duplicates.
+	EventSub struct {
+		closed   chan struct{}
+		contract *bind.BoundContract
+		eFact    EventFactory
+
+		watchLogs, filterLogs chan types.Log
+		watchSub, filterSub   event.Subscription
+	}
+
+	// Event is a generic on-chain event.
+	Event struct {
+		Name   string      // Name of the event. Must match the ABI definition.
+		Data   interface{} // Instance of the concrete Event type.
+		Filter Filter      // Filters Events by their body.
+		Log    types.Log   // Raw original log for additional information.
+	}
+
+	// EventFactory is used to create `Event`s.
+	// The `Data` and `Name` fields must be set. `Filter` is optional.
+	EventFactory func() *Event
+
+	// Filter can be used to filter events.
+	// Look at `TestEventSub_Filter` test or the auto generated Filter- and
+	// Watch-functions in the bindings/ folder for an example.
+	Filter [][]interface{}
+)
+
+// NewEventSub creates a new `EventSub`. Should always be closed with `Close`.
+// `pastBlocks` can be used to define how many blocks into the past the sub
+// should query.
+func NewEventSub(ctx context.Context, chain ethereum.ChainReader, contract *bind.BoundContract, eFact EventFactory, pastBlocks uint64) (*EventSub, error) {
+	// Get start block number.
+	startBlock, err := calcStartBlock(ctx, chain, pastBlocks)
+	if err != nil {
+		return nil, errors.WithMessage(err, "calculating starting block number")
+	}
+	// Watch for future events.
+	event := eFact()
+	watchOpts := &bind.WatchOpts{Start: &startBlock}
+	watchLogs, watchSub, err := contract.WatchLogs(watchOpts, event.Name, event.Filter...)
+	if err != nil {
+		err = cherrors.CheckIsChainNotReachableError(err)
+		return nil, errors.WithMessage(err, "watching logs")
+	}
+	// Read past events.
+	filterOpts := &bind.FilterOpts{Start: startBlock}
+	filterLogs, filterSub, err := contract.FilterLogs(filterOpts, event.Name, event.Filter...)
+	if err != nil {
+		watchSub.Unsubscribe()
+		err = cherrors.CheckIsChainNotReachableError(err)
+		return nil, errors.WithMessage(err, "filtering logs")
+	}
+
+	return &EventSub{
+		closed:     make(chan struct{}),
+		contract:   contract,
+		eFact:      eFact,
+		watchLogs:  watchLogs,
+		filterLogs: filterLogs,
+		watchSub:   watchSub,
+		filterSub:  filterSub,
+	}, nil
+}
+
+func calcStartBlock(ctx context.Context, chain ethereum.ChainReader, pastBlocks uint64) (uint64, error) {
+	current, err := chain.HeaderByNumber(ctx, nil)
+	if err != nil {
+		err = cherrors.CheckIsChainNotReachableError(err)
+		return 0, errors.WithMessage(err, "retrieving latest block")
+	}
+	if current.Number.Uint64() <= pastBlocks {
+		return 1, nil
+	}
+	return current.Number.Uint64() - pastBlocks, nil
+}
+
+// ReadAll reads all past and future events into `sink`.
+// Can be aborted by cancelling `ctx` or `Close()`.
+// It is possible that the same event is read more than once.
+func (s *EventSub) ReadAll(ctx context.Context, sink chan<- *Event) error {
+	// First read into the past.
+	if err := s.readPast(ctx, sink); err != nil {
+		return errors.WithMessage(err, "reading logs")
+	}
+	// Then wait for new events.
+	if err := s.readFuture(ctx, sink); err != nil {
+		return errors.WithMessage(err, "reading logs")
+	}
+	return nil
+}
+
+// ReadPast reads all past events into `sink`.
+// Can be aborted by cancelling `ctx` or `Close()`.
+// It is possible that the same event is read more than once.
+func (s *EventSub) ReadPast(ctx context.Context, sink chan<- *Event) error {
+	return errors.WithMessage(s.readPast(ctx, sink), "reading logs")
+}
+
+// ReadFuture reads all future events into `sink`.
+// Can be aborted by cancelling `ctx` or `Close()`.
+// It is possible that the same event is read more than once.
+func (s *EventSub) ReadFuture(ctx context.Context, sink chan<- *Event) error {
+	return errors.WithMessage(s.readFuture(ctx, sink), "reading logs")
+}
+
+func (s *EventSub) readPast(ctx context.Context, sink chan<- *Event) error {
+	var logs []types.Log
+	// Two read loops are needed if the event sub is closed before all events
+	// could be read.
+read1:
+	for {
+		select {
+		case log := <-s.filterLogs:
+			logs = append(logs, log)
+		case err := <-s.filterSub.Err():
+			if err != nil {
+				err = cherrors.CheckIsChainNotReachableError(err)
+				return err
+			}
+			break read1
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-s.closed:
+			return nil
+		}
+	}
+read2:
+	for {
+		select {
+		case log := <-s.filterLogs:
+			logs = append(logs, log)
+		case <-s.closed:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			break read2
+		}
+	}
+
+	for _, log := range logs {
+		event := s.eFact()
+		if err := s.contract.UnpackLog(event.Data, event.Name, log); err != nil {
+			return err
+		}
+		event.Log = log
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case sink <- event:
+		case <-s.closed:
+			return nil
+		}
+	}
+	return nil
+}
+
+func (s *EventSub) readFuture(ctx context.Context, sink chan<- *Event) error {
+	for {
+		select {
+		case log := <-s.watchLogs:
+			event := s.eFact()
+			if err := s.contract.UnpackLog(event.Data, event.Name, log); err != nil {
+				return err
+			}
+			event.Log = log
+
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case sink <- event:
+			case <-s.closed:
+				return nil
+			}
+		case err := <-s.watchSub.Err():
+			err = cherrors.CheckIsChainNotReachableError(err)
+			return err
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-s.closed:
+			return nil
+		}
+	}
+}
+
+// Close closes the sub and frees associated resources.
+// Should be called exactly once and panics otherwise.
+// Must not be called if the construction function returned with an error.
+func (s *EventSub) Close() {
+	close(s.closed)
+	s.watchSub.Unsubscribe()
+	s.filterSub.Unsubscribe()
+}