fix: consumer for worker and some fixes

TimeleapLabs · Apr 6, 2024 · 3611fa3 · 3611fa3
1 parent 2538aba
commit 3611fa3
Show file tree

Hide file tree

Showing 16 changed files with 146 additions and 71 deletions.
diff --git a/internal/cmd/consumer.go b/internal/cmd/consumer.go
@@ -12,6 +12,7 @@ import (
 	evmlogService "github.com/KenshiTech/unchained/service/evmlog"
 	uniswapService "github.com/KenshiTech/unchained/service/uniswap"
 	"github.com/KenshiTech/unchained/transport/client"
+	"github.com/KenshiTech/unchained/transport/client/conn"
 	"github.com/KenshiTech/unchained/transport/client/handler"
 	"github.com/KenshiTech/unchained/transport/server"
 	"github.com/KenshiTech/unchained/transport/server/gql"
@@ -25,6 +26,7 @@ var consumerCmd = &cobra.Command{
 
 	PreRun: func(cmd *cobra.Command, args []string) {
 		config.App.Network.BrokerURI = cmd.Flags().Lookup("broker").Value.String()
+		config.App.Network.Bind = cmd.Flags().Lookup("graphql").Value.String()
 	},
 
 	Run: func(cmd *cobra.Command, args []string) {
@@ -45,16 +47,18 @@ var consumerCmd = &cobra.Command{
 		pos := pos.New(ethRPC)
 		db.Start()
 
-		server.New(
-			gql.WithGraphQL(),
-		)
-
 		correctnessService := correctnessService.New(ethRPC)
 		evmLogService := evmlogService.New(ethRPC, pos)
 		uniswapService := uniswapService.New(ethRPC, pos)
 
-		handler := handler.New(correctnessService, uniswapService, evmLogService)
+		conn.Start()
+
+		handler := handler.NewConsumerHandler(correctnessService, uniswapService, evmLogService)
 		client.Consume(handler)
+
+		server.New(
+			gql.WithGraphQL(),
+		)
 	},
 }
 
@@ -67,4 +71,11 @@ func init() {
 		"wss://shinobi.brokers.kenshi.io",
 		"Unchained broker to connect to",
 	)
+
+	consumerCmd.Flags().StringP(
+		"graphql",
+		"g",
+		"127.0.0.1:8080",
+		"The graphql server path to bind",
+	)
 }
diff --git a/internal/cmd/worker.go b/internal/cmd/worker.go
@@ -9,10 +9,10 @@ import (
 	"github.com/KenshiTech/unchained/persistence"
 	"github.com/KenshiTech/unchained/pos"
 	"github.com/KenshiTech/unchained/scheduler"
-	correctnessService "github.com/KenshiTech/unchained/service/correctness"
 	evmlogService "github.com/KenshiTech/unchained/service/evmlog"
 	uniswapService "github.com/KenshiTech/unchained/service/uniswap"
 	"github.com/KenshiTech/unchained/transport/client"
+	"github.com/KenshiTech/unchained/transport/client/conn"
 	"github.com/KenshiTech/unchained/transport/client/handler"
 	"github.com/spf13/cobra"
 )
@@ -45,17 +45,20 @@ var workerCmd = &cobra.Command{
 		pos := pos.New(ethRPC)
 		badger := persistence.New(contextPath)
 
-		correctnessService := correctnessService.New(ethRPC)
 		evmLogService := evmlogService.New(ethRPC, pos)
 		uniswapService := uniswapService.New(ethRPC, pos)
 
-		scheduler.New(
+		scheduler := scheduler.New(
 			scheduler.WithEthLogs(evmLogService, ethRPC, badger),
 			scheduler.WithUniswapEvents(uniswapService, ethRPC),
 		)
 
-		handler := handler.New(correctnessService, uniswapService, evmLogService)
+		conn.Start()
+
+		handler := handler.NewWorkerHandler()
 		client.Consume(handler)
+
+		scheduler.Start()
 	},
 }
 

diff --git a/internal/scheduler/logs/logs.go b/internal/scheduler/logs/logs.go
@@ -40,8 +40,6 @@ func (e *EvmLog) Run() {
 		return
 	}
 
-	log.Logger.With("Chain", e.chain).Info("Run evm log task")
-
 	for _, conf := range config.App.Plugins.EthLog.Events {
 		if conf.Chain != e.chain {
 			continue

diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
@@ -95,4 +95,6 @@ func (s *Scheduler) AddTask(duration time.Duration, task Task) {
 
 func (s *Scheduler) Start() {
 	s.scheduler.Start()
+
+	select {}
 }
diff --git a/internal/scheduler/uniswap/uniswap.go b/internal/scheduler/uniswap/uniswap.go
@@ -30,8 +30,6 @@ func (u *Uniswap) Run() {
 		return
 	}
 
-	log.Logger.With("Chain", u.chain).Info("Run Uniswap task")
-
 	currBlockNumber, err := u.uniswapService.GetBlockNumber(u.chain)
 	if err != nil {
 		log.Logger.Error(

diff --git a/internal/service/uniswap/uniswap.go b/internal/service/uniswap/uniswap.go
@@ -7,6 +7,9 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"time"
+
+	"github.com/KenshiTech/unchained/utils"
 
 	"github.com/KenshiTech/unchained/config"
 
@@ -567,11 +570,20 @@ func New(ethRPC *ethereum.Repository, pos *pos.Repository) *Service {
 		ethRPC: ethRPC,
 		pos:    pos,
 
-		crossPrices:     map[string]big.Int{},
-		crossTokens:     map[string]datasets.TokenKey{},
+		consensus:       nil,
+		signatureCache:  nil,
+		aggregateCache:  nil,
+		signatureMutex:  sync.Mutex{},
+		LastBlock:       *xsync.NewMapOf[datasets.TokenKey, uint64](),
 		SupportedTokens: map[datasets.TokenKey]bool{},
 		PriceCache:      map[string]*lru.Cache[uint64, big.Int]{},
+		crossPrices:     map[string]big.Int{},
+		crossTokens:     map[string]datasets.TokenKey{},
 	}
+	DebouncedSaveSignatures = utils.Debounce[datasets.AssetKey, SaveSignatureArgs](5*time.Second, u.saveSignatures)
+	u.twoOneNineTwo.Exp(big.NewInt(2), big.NewInt(192), nil)
+	u.tenEighteen.Exp(big.NewInt(10), big.NewInt(18), nil)
+	u.tenEighteenF.SetInt(&u.tenEighteen)
 
 	if config.App.Plugins.Uniswap != nil {
 		for _, t := range config.App.Plugins.Uniswap.Tokens {
@@ -582,10 +594,6 @@ func New(ethRPC *ethereum.Repository, pos *pos.Repository) *Service {
 		}
 	}
 
-	u.twoOneNineTwo.Exp(big.NewInt(2), big.NewInt(192), nil)
-	u.tenEighteen.Exp(big.NewInt(10), big.NewInt(18), nil)
-	u.tenEighteenF.SetInt(&u.tenEighteen)
-
 	var err error
 	u.signatureCache, err = lru.New[bls12381.G1Affine, []datasets.Signature](evmlog.LruSize)
 	if err != nil {

diff --git a/internal/transport/client/client.go b/internal/transport/client/client.go
@@ -7,37 +7,42 @@ import (
 	"github.com/KenshiTech/unchained/transport/client/handler"
 )
 
-func Consume(handler *handler.Handler) {
-	conn.Start()
-
+func Consume(handler handler.Handler) {
 	incoming := conn.Read()
 
-	for payload := range incoming {
-		switch opcodes.OpCode(payload[0]) {
-		case opcodes.Feedback:
-			log.Logger.
-				With("Feedback", string(payload[1:])).
-				Info("Broker")
+	go func() {
+		log.Logger.Info("Starting consumer from broker")
 
-		case opcodes.KoskChallenge:
-			challenge := handler.Challenge(payload[1:])
-			conn.Send(opcodes.KoskResult, challenge.Sia().Content)
+		for payload := range incoming {
+			switch opcodes.OpCode(payload[0]) {
+			case opcodes.Error:
+				log.Logger.
+					With("Error", string(payload[1:])).
+					Error("Broker")
 
-		case opcodes.PriceReportBroadcast:
-			go handler.PriceReport(payload[1:])
+			case opcodes.Feedback:
+				log.Logger.
+					With("Feedback", string(payload[1:])).
+					Warn("Broker")
 
-		case opcodes.EventLogBroadcast:
-			go handler.EventLog(payload[1:])
+			case opcodes.KoskChallenge:
+				challenge := handler.Challenge(payload[1:])
+				conn.Send(opcodes.KoskResult, challenge.Sia().Content)
 
-		case opcodes.CorrectnessReportBroadcast:
-			go handler.CorrectnessReport(payload[1:])
+			case opcodes.PriceReportBroadcast:
+				go handler.PriceReport(payload[1:])
 
-		default:
-			log.Logger.
-				With("Code", payload[0]).
-				Info("Unknown call code")
-		}
-	}
+			case opcodes.EventLogBroadcast:
+				go handler.EventLog(payload[1:])
 
-	log.Logger.Error("Client loop breaks")
+			case opcodes.CorrectnessReportBroadcast:
+				go handler.CorrectnessReport(payload[1:])
+
+			default:
+				log.Logger.
+					With("Code", payload[0]).
+					Info("Unknown call code")
+			}
+		}
+	}()
 }
diff --git a/internal/transport/client/conn/conn.go b/internal/transport/client/conn/conn.go
@@ -35,7 +35,6 @@ func Start() {
 	}
 
 	Send(opcodes.Hello, bls.ClientSigner.Sia().Content)
-	Send(opcodes.RegisterConsumer, nil)
 }
 
 func Reconnect(err error) {
@@ -119,6 +118,8 @@ func Read() <-chan []byte {
 
 				continue
 			}
+
+			out <- payload
 		}
 	}()
 

diff --git a/internal/transport/client/handler/challenge.go b/internal/transport/client/handler/challenge.go
@@ -6,7 +6,16 @@ import (
 	sia "github.com/pouya-eghbali/go-sia/v2/pkg"
 )
 
-func (h *Handler) Challenge(message []byte) *kosk.Challenge {
+func (h *consumer) Challenge(message []byte) *kosk.Challenge {
+	challenge := new(kosk.Challenge).DeSia(&sia.Sia{Content: message})
+
+	signature, _ := bls.Sign(*bls.ClientSecretKey, challenge.Random[:])
+	challenge.Signature = signature.Bytes()
+
+	return challenge
+}
+
+func (w worker) Challenge(message []byte) *kosk.Challenge {
 	challenge := new(kosk.Challenge).DeSia(&sia.Sia{Content: message})
 
 	signature, _ := bls.Sign(*bls.ClientSecretKey, challenge.Random[:])

diff --git a/internal/transport/client/handler/consumer.go b/internal/transport/client/handler/consumer.go
@@ -0,0 +1,29 @@
+package handler
+
+import (
+	"github.com/KenshiTech/unchained/constants/opcodes"
+	"github.com/KenshiTech/unchained/service/correctness"
+	"github.com/KenshiTech/unchained/service/evmlog"
+	"github.com/KenshiTech/unchained/service/uniswap"
+	"github.com/KenshiTech/unchained/transport/client/conn"
+)
+
+type consumer struct {
+	correctness *correctness.Service
+	uniswap     *uniswap.Service
+	evmlog      *evmlog.Service
+}
+
+func NewConsumerHandler(
+	correctness *correctness.Service,
+	uniswap *uniswap.Service,
+	evmlog *evmlog.Service,
+) Handler {
+	conn.Send(opcodes.RegisterConsumer, nil)
+
+	return &consumer{
+		correctness: correctness,
+		uniswap:     uniswap,
+		evmlog:      evmlog,
+	}
+}
diff --git a/internal/transport/client/handler/correctness.go b/internal/transport/client/handler/correctness.go
@@ -7,7 +7,7 @@ import (
 	sia "github.com/pouya-eghbali/go-sia/v2/pkg"
 )
 
-func (h *Handler) CorrectnessReport(message []byte) {
+func (h *consumer) CorrectnessReport(message []byte) {
 	packet := new(datasets.BroadcastCorrectnessPacket).DeSia(&sia.Sia{Content: message})
 	toHash := packet.Info.Sia().Content
 	hash, err := bls.Hash(toHash)
@@ -38,3 +38,5 @@ func (h *Handler) CorrectnessReport(message []byte) {
 		true,
 	)
 }
+
+func (w worker) CorrectnessReport(message []byte) {}
diff --git a/internal/transport/client/handler/event.go b/internal/transport/client/handler/event.go
@@ -7,7 +7,7 @@ import (
 	sia "github.com/pouya-eghbali/go-sia/v2/pkg"
 )
 
-func (h *Handler) EventLog(message []byte) {
+func (h *consumer) EventLog(message []byte) {
 	packet := new(datasets.BroadcastEventPacket).DeSia(&sia.Sia{Content: message})
 	toHash := packet.Info.Sia().Content
 	hash, err := bls.Hash(toHash)
@@ -39,3 +39,5 @@ func (h *Handler) EventLog(message []byte) {
 		false,
 	)
 }
+
+func (w worker) EventLog(message []byte) {}
diff --git a/internal/transport/client/handler/handler.go b/internal/transport/client/handler/handler.go
@@ -1,25 +1,12 @@
 package handler
 
 import (
-	"github.com/KenshiTech/unchained/service/correctness"
-	"github.com/KenshiTech/unchained/service/evmlog"
-	"github.com/KenshiTech/unchained/service/uniswap"
+	"github.com/KenshiTech/unchained/crypto/kosk"
 )
 
-type Handler struct {
-	correctness *correctness.Service
-	uniswap     *uniswap.Service
-	evmlog      *evmlog.Service
-}
-
-func New(
-	correctness *correctness.Service,
-	uniswap *uniswap.Service,
-	evmlog *evmlog.Service,
-) *Handler {
-	return &Handler{
-		correctness: correctness,
-		uniswap:     uniswap,
-		evmlog:      evmlog,
-	}
+type Handler interface {
+	Challenge(message []byte) *kosk.Challenge
+	CorrectnessReport(message []byte)
+	EventLog(message []byte)
+	PriceReport(message []byte)
 }
diff --git a/internal/transport/client/handler/price.go b/internal/transport/client/handler/price.go
@@ -7,7 +7,7 @@ import (
 	sia "github.com/pouya-eghbali/go-sia/v2/pkg"
 )
 
-func (h *Handler) PriceReport(message []byte) {
+func (h *consumer) PriceReport(message []byte) {
 	packet := new(datasets.BroadcastPricePacket).DeSia(&sia.Sia{Content: message})
 	toHash := packet.Info.Sia().Content
 	hash, err := bls.Hash(toHash)
@@ -39,3 +39,5 @@ func (h *Handler) PriceReport(message []byte) {
 		false,
 	)
 }
+
+func (w worker) PriceReport(message []byte) {}
diff --git a/internal/transport/client/handler/worker.go b/internal/transport/client/handler/worker.go
@@ -0,0 +1,8 @@
+package handler
+
+type worker struct {
+}
+
+func NewWorkerHandler() Handler {
+	return &worker{}
+}