feat(claimer): rewrite claimer in go

.gitignore

@@ -12,3 +12,4 @@ machine-snapshot/**
 /cartesi-rollups-node
 /cartesi-rollups-advancer
 /cartesi-rollups-validator
+/cartesi-rollups-claimer

Makefile

@@ -34,7 +34,7 @@ CLAIMER := cmd/authority-claimer/target/$(BUILD_TYPE)/cartesi-rollups-authority-
 RUST_ARTIFACTS := $(CLAIMER)
 
 # Go artifacts
-GO_ARTIFACTS := cartesi-rollups-node cartesi-rollups-cli cartesi-rollups-evm-reader cartesi-rollups-advancer cartesi-rollups-validator
+GO_ARTIFACTS := cartesi-rollups-node cartesi-rollups-cli cartesi-rollups-evm-reader cartesi-rollups-advancer cartesi-rollups-validator cartesi-rollups-claimer
 
 # fixme(vfusco): path on all oses
 CGO_CFLAGS:= -I$(PREFIX)/include

cmd/cartesi-rollups-claimer/main.go

@@ -0,0 +1,16 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package main
+
+import (
+	"github.com/cartesi/rollups-node/cmd/cartesi-rollups-claimer/root"
+	"os"
+)
+
+func main() {
+	err := root.Cmd.Execute()
+	if err != nil {
+		os.Exit(1)
+	}
+}

cmd/cartesi-rollups-claimer/root/root.go

@@ -0,0 +1,70 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package root
+
+import (
+	"log/slog"
+
+	"github.com/cartesi/rollups-node/internal/claimer"
+	"github.com/cartesi/rollups-node/internal/config"
+	"github.com/cartesi/rollups-node/pkg/service"
+	"github.com/spf13/cobra"
+)
+
+var (
+	// Should be overridden during the final release build with ldflags
+	// to contain the actual version number
+	buildVersion   = "devel"
+	claimerService = claimer.Service{}
+	createInfo     = claimer.CreateInfo{
+		CreateInfo: service.CreateInfo{
+			Name:                 "claimer",
+			ProcOwner:            true,
+			EnableSignalHandling: true,
+			TelemetryCreate:      true,
+			TelemetryAddress:     ":8081",
+			Impl:                 &claimerService,
+		},
+	}
+)
+
+var Cmd = &cobra.Command{
+	Use:   createInfo.Name,
+	Short: "Runs " + createInfo.Name,
+	Long:  "Runs " + createInfo.Name + " in standalone mode",
+	Run:   run,
+}
+
+func init() {
+	c := config.FromEnv()
+	createInfo.Auth = c.Auth
+	createInfo.BlockchainHttpEndpoint = c.BlockchainHttpEndpoint
+	createInfo.PostgresEndpoint = c.PostgresEndpoint
+	createInfo.PollInterval = c.ClaimerPollingInterval
+	createInfo.LogLevel = map[slog.Level]string{
+		slog.LevelDebug: "debug",
+		slog.LevelInfo:  "info",
+		slog.LevelWarn:  "warn",
+		slog.LevelError: "error",
+	}[c.LogLevel]
+
+	Cmd.Flags().StringVar(&createInfo.TelemetryAddress,
+		"telemetry-address", createInfo.TelemetryAddress,
+		"health check and metrics address and port")
+	Cmd.Flags().StringVar(&createInfo.BlockchainHttpEndpoint.Value,
+		"blockchain-http-endpoint", createInfo.BlockchainHttpEndpoint.Value,
+		"blockchain http endpoint")
+	Cmd.Flags().DurationVar(&createInfo.PollInterval,
+		"poll-interval", createInfo.PollInterval,
+		"poll interval")
+	Cmd.Flags().StringVar(&createInfo.LogLevel,
+		"log-level", createInfo.LogLevel,
+		"log level: debug, info, warn, error.")
+}
+
+func run(cmd *cobra.Command, args []string) {
+	cobra.CheckErr(claimer.Create(createInfo, &claimerService))
+	claimerService.CreateDefaultHandlers("/" + claimerService.Name)
+	cobra.CheckErr(claimerService.Serve())
+}

internal/claimer/auth.go

@@ -0,0 +1,68 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package claimer
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cartesi/rollups-node/internal/config"
+	signtx "github.com/cartesi/rollups-node/internal/kms"
+	"github.com/cartesi/rollups-node/pkg/ethutil"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethclient"
+
+	aws "github.com/aws/aws-sdk-go-v2/aws"
+	aws_cfg "github.com/aws/aws-sdk-go-v2/config"
+	aws_kms "github.com/aws/aws-sdk-go-v2/service/kms"
+)
+
+var (
+	ENoAuth = fmt.Errorf("error: unimplemented authentication method")
+)
+
+func CreateSignerFromAuth(
+	auth config.Auth,
+	ctx context.Context,
+	client *ethclient.Client,
+) (
+	*bind.TransactOpts,
+	error,
+) {
+	chainID, err := client.ChainID(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	switch auth := auth.(type) {
+	case config.AuthPrivateKey:
+		key, err := crypto.HexToECDSA(auth.PrivateKey.Value)
+		if err != nil {
+			return nil, err
+		}
+		return bind.NewKeyedTransactorWithChainID(key, chainID)
+	case config.AuthMnemonic:
+		privateKey, err := ethutil.MnemonicToPrivateKey(
+			auth.Mnemonic.Value, uint32(auth.AccountIndex.Value))
+		if err != nil {
+			return nil, err
+		}
+		return bind.NewKeyedTransactorWithChainID(privateKey, chainID)
+	case config.AuthAWS:
+		awsc, err := aws_cfg.LoadDefaultConfig(ctx)
+		if err != nil {
+			return nil, err
+		}
+		kms := aws_kms.NewFromConfig(awsc)
+		// TODO: option for an alternative endpoint
+		//kms := aws_kms.NewFromConfig(awsc, func(o *aws_kms.Options) {
+		//	o.BaseEndpoint = aws.String(auth.EndpointURL.Value)
+		//})
+		return signtx.CreateAWSTransactOpts(ctx, kms,
+			aws.String(auth.KeyID.Value), types.NewEIP155Signer(chainID))
+	}
+	return nil, ENoAuth
+}

internal/claimer/claimer.go

@@ -0,0 +1,204 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package claimer
+
+import (
+	"context"
+
+	. "github.com/cartesi/rollups-node/internal/config"
+	. "github.com/cartesi/rollups-node/internal/repository"
+	"github.com/cartesi/rollups-node/pkg/service"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	. "github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+type CreateInfo struct {
+	service.CreateInfo
+
+	Auth   Auth
+	Signer *bind.TransactOpts
+
+	BlockchainHttpEndpoint Redacted[string]
+	EthConn                *ethclient.Client
+
+	PostgresEndpoint Redacted[string]
+	DBConn           *Database
+}
+
+type Service struct {
+	service.Service
+
+	DBConn         *Database
+	EthConn        *ethclient.Client
+	Signer         *bind.TransactOpts
+	ClaimsInFlight map[Hash]Hash // claimHash -> txHash
+}
+
+func Create(ci CreateInfo, s *Service) error {
+	var err error
+
+	err = service.Create(&ci.CreateInfo, &s.Service)
+	if err != nil {
+		return err
+	}
+
+	if s.EthConn == nil {
+		if ci.EthConn == nil {
+			ci.EthConn, err = ethclient.Dial(ci.BlockchainHttpEndpoint.Value)
+			if err != nil {
+				return err
+			}
+		}
+		s.EthConn = ci.EthConn
+	}
+
+	if s.DBConn == nil {
+		if ci.DBConn == nil {
+			ci.DBConn, err = Connect(s.Context, ci.PostgresEndpoint.Value)
+			if err != nil {
+				return err
+			}
+		}
+		s.DBConn = ci.DBConn
+	}
+
+	if s.ClaimsInFlight == nil {
+		s.ClaimsInFlight = map[Hash]Hash{}
+	}
+
+	if s.Signer == nil {
+		if ci.Signer == nil {
+			ci.Signer, err = CreateSignerFromAuth(ci.Auth, s.Context, s.EthConn)
+			if err != nil {
+				return err
+			}
+			s.Signer = ci.Signer
+		}
+	}
+
+	return err
+}
+
+func (s *Service) Alive() bool {
+	return true
+}
+
+func (s *Service) Ready() bool {
+	return true
+}
+
+func (s *Service) Reload() []error {
+	return nil
+}
+
+func (s *Service) Stop(bool) []error {
+	return nil
+}
+
+func (s *Service) Tick() []error {
+	err := s.submitClaimsAndUpdateDatabase(s)
+	if err != nil {
+		return []error{err}
+	}
+	return nil
+}
+
+func (s *Service) submitClaimsAndUpdateDatabase(se SideEffects) error {
+	claims, err := se.selectComputedClaims()
+	if err != nil {
+		return err
+	}
+
+	claimFromHash := make(map[Hash]*ComputedClaim)
+	for i := 0; i < len(claims); i++ {
+		claimFromHash[claims[i].Hash] = &claims[i]
+	}
+
+	// check claims in flight
+	for claimHash, txHash := range s.ClaimsInFlight {
+		ready, receipt, err := se.pollTransaction(txHash)
+		if err != nil {
+			return err
+		}
+		if !ready {
+			continue
+		}
+
+		if claim, ok := claimFromHash[claimHash]; ok {
+			err = se.updateEpochWithSubmittedClaim(
+				s.DBConn,
+				s.Context,
+				claim,
+				receipt.TxHash)
+			if err != nil {
+				return err
+			}
+			delete(s.ClaimsInFlight, claimHash)
+			delete(claimFromHash, claimHash)
+			s.Logger.Info("claimer: Claim submitted",
+				"app", claim.AppContractAddress,
+				"claim", claimHash,
+				"last_block", claim.EpochLastBlock,
+				"tx", receipt.TxHash)
+		}
+	}
+
+	// check event logs for the remaining claims, submit if not found
+	for i := 0; i < len(claims); i++ {
+		_, isSelected := claimFromHash[claims[i].Hash]
+		_, isInFlight := s.ClaimsInFlight[claims[i].Hash]
+		if !isSelected || isInFlight {
+			continue
+		}
+
+		it, inst, err := se.enumerateSubmitClaimEventsSince(
+			s.EthConn, s.Context,
+			claims[i].AppIConsensusAddress,
+			claims[i].EpochLastBlock)
+		if err != nil {
+			return err
+		}
+
+		for event, err := range it {
+			if err != nil {
+				return err
+			}
+
+			hash := Hash(event.Claim)
+			claim, ok := claimFromHash[hash]
+			if ok {
+				err := se.updateEpochWithSubmittedClaim(
+					s.DBConn,
+					s.Context,
+					claim,
+					event.Raw.TxHash)
+				if err != nil {
+					return err
+				}
+				delete(claimFromHash, hash)
+			}
+		}
+
+		// submit if not found in the logs (fetch from hash again, can be stale)
+		if claim, ok := claimFromHash[claims[i].Hash]; ok {
+			txHash, err := se.submitClaimToBlockchain(inst, s.Signer, claim)
+			if err != nil {
+				return err
+			}
+			s.ClaimsInFlight[claims[i].Hash] = txHash
+			delete(claimFromHash, claim.Hash)
+		}
+	}
+	return nil
+}
+
+func (s *Service) Start(context context.Context, ready chan<- struct{}) error {
+	ready <- struct{}{}
+	return s.Serve()
+}
+func (s *Service) String() string {
+	return s.Name
+}