Merge branch 'main' into walkah/fix-separate-bacalhau-container

.gitignore

@@ -20,3 +20,4 @@ bin/
 vendor/
 /data/
 .DS_Store
+/chain-data/*

pkg/solver/controller.go

@@ -351,17 +351,23 @@ func (controller *SolverController) addResourceOffer(resourceOffer data.Resource
 
 	// If the balance is less than the required balance, don't add the resource offer
 	if balance.Cmp(requiredBalanceWei) < 0 {
-		return nil, fmt.Errorf("address %s doesn't have enough ETH balance. The required balance is %s but current balance is %s", resourceOffer.ResourceProvider, requiredBalanceWei, balance)
+		err := fmt.Errorf("address %s doesn't have enough ETH balance. The required balance is %s but current balance is %s", resourceOffer.ResourceProvider, requiredBalanceWei, balance)
+		controller.log.Error("ETH balance check failed", err)
+		return nil, nil
 	}
 
 	// required LP balance
 	requiredBalanceLp := web3.EtherToWei(float64(resourceOffer.DefaultPricing.InstructionPrice)) // based on the required LP balance for a job
 	balanceLp, err := controller.web3SDK.GetLPBalance(resourceOffer.ResourceProvider)
 	if err != nil {
-		return nil, fmt.Errorf("failed to retrieve LP balance for resource provider: %v", err)
+		err := fmt.Errorf("failed to retrieve LP balance for resource provider: %v", err)
+		controller.log.Error("LP Balance error", err)
+		return nil, nil
 	}
 	if balanceLp.Cmp(requiredBalanceLp) < 0 {
-		return nil, fmt.Errorf("address %s doesn't have enough LP balance. The required balance is %s but current balance is %s", resourceOffer.ResourceProvider, requiredBalanceLp, balanceLp)
+		err := fmt.Errorf("address %s doesn't have enough LP balance. The required balance is %s but current balance is %s", resourceOffer.ResourceProvider, requiredBalanceLp, balanceLp)
+		controller.log.Error("LP balance check failed", err)
+		return nil, nil
 	}
 
 	controller.log.Info("add resource offer", resourceOffer)

pkg/solver/server.go

@@ -69,7 +69,7 @@ func (solverServer *solverServer) ListenAndServe(ctx context.Context, cm *system
 	subrouter.Use(httprate.Limit(
 		solverServer.options.RateLimiter.RequestLimit,
 		time.Duration(solverServer.options.RateLimiter.WindowLength)*time.Second,
-		httprate.WithKeyFuncs(httprate.KeyByIP, httprate.KeyByEndpoint),
+		httprate.WithKeyFuncs(httprate.KeyByRealIP, httprate.KeyByEndpoint),
 	))
 
 	subrouter.HandleFunc("/job_offers", http.GetHandler(solverServer.getJobOffers)).Methods("GET")

pkg/web3/events_jobcreator.go

@@ -49,12 +49,7 @@ func (s *JobCreatorEventChannels) Start(
 	if err != nil {
 		return err
 	}
-
-	defer func() {
-		if jobAddedSub != nil {
-			jobAddedSub.Unsubscribe()
-		}
-	}()
+	cm.RegisterCallback(unsubscribeSub(jobAddedSub))
 
 	for {
 		select {

pkg/web3/events_mediation.go

@@ -49,12 +49,7 @@ func (m *MediationEventChannels) Start(
 	if err != nil {
 		return err
 	}
-
-	defer func() {
-		if mediationRequestedSub != nil {
-			mediationRequestedSub.Unsubscribe()
-		}
-	}()
+	cm.RegisterCallback(unsubscribeSub(mediationRequestedSub))
 
 	for {
 		select {

pkg/web3/events_payments.go

@@ -49,12 +49,8 @@ func (p *PaymentEventChannels) Start(
 	if err != nil {
 		return err
 	}
+	cm.RegisterCallback(unsubscribeSub(paymentSub))
 
-	defer func() {
-		if paymentSub != nil {
-			paymentSub.Unsubscribe()
-		}
-	}()
 	for {
 		select {
 		case <-ctx.Done():

pkg/web3/events_pow.go

@@ -3,6 +3,7 @@ package web3
 import (
 	"context"
 	"fmt"
+
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/lilypad-tech/lilypad/pkg/system"
@@ -49,12 +50,7 @@ func (s *PowEventChannels) Start(
 	if err != nil {
 		return err
 	}
-
-	defer func() {
-		if newPowRoundSub != nil {
-			newPowRoundSub.Unsubscribe()
-		}
-	}()
+	cm.RegisterCallback(unsubscribeSub(newPowRoundSub))
 
 	for {
 		select {

pkg/web3/events_storage.go

@@ -49,12 +49,7 @@ func (s *StorageEventChannels) Start(
 	if err != nil {
 		return err
 	}
-
-	defer func() {
-		if dealStateChangeSub != nil {
-			dealStateChangeSub.Unsubscribe()
-		}
-	}()
+	cm.RegisterCallback(unsubscribeSub(dealStateChangeSub))
 
 	for {
 		select {

pkg/web3/events_token.go

@@ -52,12 +52,7 @@ func (t *TokenEventChannels) Start(
 	if err != nil {
 		return err
 	}
-
-	defer func() {
-		if transferSub != nil {
-			transferSub.Unsubscribe()
-		}
-	}()
+	cm.RegisterCallback(unsubscribeSub(transferSub))
 
 	for {
 		select {

pkg/web3/utils.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/event"
 )
 
 func GetPublicKey(privateKey *ecdsa.PrivateKey) ecdsa.PublicKey {
@@ -68,3 +69,10 @@ func ConvertStringToInt64(st string) uint64 {
 	bigInt := ConvertStringToBigInt(st)
 	return bigInt.Uint64()
 }
+
+func unsubscribeSub(sub event.Subscription) func() error {
+	return func() error {
+		sub.Unsubscribe()
+		return nil
+	}
+}