[scheduler] support restart & retry (#422)

Author: huangzhiran

cmd/coordinator/config/config.go

@@ -33,7 +33,7 @@ var (
 		DatabaseDSN:            "postgres://test_user:test_passwd@postgres:5432/test?sslmode=disable",
 		BootNodeMultiAddr:      "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
 		IoTeXChainID:           2,
-		ProjectContractAddress: "0x2339644f65c371Ca36b335A7eC3EB8AD8CBd5F51",
+		ProjectContractAddress: "0x57872cF4b6D02053D5486F6882062c41a0F5C26a",
 		IPFSEndpoint:           "ipfs.mainnet.iotex.io",
 		DIDAuthServerEndpoint:  "didkit:9999",
 		SequencerPubKey:        "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
@@ -46,9 +46,8 @@ var (
 		DatabaseDSN:            "postgres://test_user:test_passwd@localhost:5432/test?sslmode=disable",
 		BootNodeMultiAddr:      "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
 		IoTeXChainID:           2,
-		ProjectContractAddress: "0x2339644f65c371Ca36b335A7eC3EB8AD8CBd5F51",
+		ProjectContractAddress: "0x57872cF4b6D02053D5486F6882062c41a0F5C26a",
 		IPFSEndpoint:           "ipfs.mainnet.iotex.io",
-		ProjectFileDirectory:   "../../test/project",
 		DIDAuthServerEndpoint:  "localhost:9999",
 		ProjectCacheDirectory:  "./project_cache",
 		SequencerPubKey:        "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",

cmd/coordinator/main.go

@@ -33,7 +33,7 @@ func main() {
 	projectManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,
 		conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)
 	if err != nil {
-		log.Fatal(errors.Wrap(err, "failed to new project config manager"))
+		log.Fatal(errors.Wrap(err, "failed to new project manager"))
 	}
 
 	if err := task.RunDispatcher(persistence, datasource.NewPostgres, projectManager.GetCachedProjectIDs, projectManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519, conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {

cmd/prover/config/config.go

@@ -8,82 +8,81 @@ import (
 )
 
 type Config struct {
-	Risc0ServerEndpoint    string `env:"RISC0_SERVER_ENDPOINT"`
-	Halo2ServerEndpoint    string `env:"HALO2_SERVER_ENDPOINT"`
-	ZKWasmServerEndpoint   string `env:"ZKWASM_SERVER_ENDPOINT"`
-	WasmServerEndpoint     string `env:"WASM_SERVER_ENDPOINT"`
-	ChainEndpoint          string `env:"CHAIN_ENDPOINT"`
-	ProjectContractAddress string `env:"PROJECT_CONTRACT_ADDRESS,optional"`
-	DatabaseDSN            string `env:"DATABASE_DSN"`
-	BootNodeMultiAddr      string `env:"BOOTNODE_MULTIADDR"`
-	ProverContractAddress  string `env:"PROVER_CONTRACT_ADDRESS,optional"`
-	ProverPrivateKey       string `env:"PROVER_PRIVATE_KEY,optional"`
-	IoTeXChainID           int    `env:"IOTEX_CHAINID"`
-	SchedulerEpoch         uint64 `env:"SCHEDULER_EPOCH"`
-	IPFSEndpoint           string `env:"IPFS_ENDPOINT"`
-	ProjectFileDirectory   string `env:"PROJECT_FILE_DIRECTORY,optional"`
-	ProjectCacheDirectory  string `env:"PROJECT_CACHE_DIRECTORY,optional"`
-	LogLevel               int    `env:"LOG_LEVEL,optional"`
-	SequencerPubKey        string `env:"SEQUENCER_PUBKEY,optional"`
-	env                    string `env:"-"`
+	Risc0ServerEndpoint      string `env:"RISC0_SERVER_ENDPOINT"`
+	Halo2ServerEndpoint      string `env:"HALO2_SERVER_ENDPOINT"`
+	ZKWasmServerEndpoint     string `env:"ZKWASM_SERVER_ENDPOINT"`
+	WasmServerEndpoint       string `env:"WASM_SERVER_ENDPOINT"`
+	ChainEndpoint            string `env:"CHAIN_ENDPOINT"`
+	ProjectContractAddress   string `env:"PROJECT_CONTRACT_ADDRESS,optional"`
+	DatabaseDSN              string `env:"DATABASE_DSN"`
+	BootNodeMultiAddr        string `env:"BOOTNODE_MULTIADDR"`
+	ProverContractAddress    string `env:"PROVER_CONTRACT_ADDRESS,optional"`
+	ProverOperatorPrivateKey string `env:"PROVER_OPERATOR_PRIVATE_KEY,optional"`
+	IoTeXChainID             int    `env:"IOTEX_CHAINID"`
+	SchedulerEpoch           uint64 `env:"SCHEDULER_EPOCH"`
+	IPFSEndpoint             string `env:"IPFS_ENDPOINT"`
+	ProjectFileDirectory     string `env:"PROJECT_FILE_DIRECTORY,optional"`
+	ProjectCacheDirectory    string `env:"PROJECT_CACHE_DIRECTORY,optional"`
+	LogLevel                 int    `env:"LOG_LEVEL,optional"`
+	SequencerPubKey          string `env:"SEQUENCER_PUBKEY,optional"`
+	env                      string `env:"-"`
 }
 
 var (
 	defaultConfig = &Config{
-		Risc0ServerEndpoint:    "risc0:4001",
-		Halo2ServerEndpoint:    "halo2:4001",
-		ZKWasmServerEndpoint:   "zkwasm:4001",
-		WasmServerEndpoint:     "wasm:4001",
-		ChainEndpoint:          "https://babel-api.testnet.iotex.io",
-		ProjectContractAddress: "0x2339644f65c371Ca36b335A7eC3EB8AD8CBd5F51",
-		DatabaseDSN:            "postgres://test_user:test_passwd@postgres:5432/test?sslmode=disable",
-		BootNodeMultiAddr:      "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
-		ProverContractAddress:  "0x2B9BC8eC74E7F2526045eb13fFa37b10e0d40464",
-		ProverPrivateKey:       "did:key:z6MkmF1AgufHf8ASaxDcCR8iSZjEsEbJMp7LkqyEHw6123",
-		IoTeXChainID:           2,
-		SchedulerEpoch:         20,
-		IPFSEndpoint:           "ipfs.mainnet.iotex.io",
-		LogLevel:               int(slog.LevelDebug),
-		SequencerPubKey:        "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
+		Risc0ServerEndpoint:      "risc0:4001",
+		Halo2ServerEndpoint:      "halo2:4001",
+		ZKWasmServerEndpoint:     "zkwasm:4001",
+		WasmServerEndpoint:       "wasm:4001",
+		ChainEndpoint:            "https://babel-api.testnet.iotex.io",
+		ProjectContractAddress:   "0x57872cF4b6D02053D5486F6882062c41a0F5C26a",
+		DatabaseDSN:              "postgres://test_user:test_passwd@postgres:5432/test?sslmode=disable",
+		BootNodeMultiAddr:        "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
+		ProverContractAddress:    "0x5B1AE92ed4a819C0B0e957a67F8dc39c0aC17Dfc",
+		ProverOperatorPrivateKey: "did:key:z6MkmF1AgufHf8ASaxDcCR8iSZjEsEbJMp7LkqyEHw6123",
+		IoTeXChainID:             2,
+		SchedulerEpoch:           20,
+		IPFSEndpoint:             "ipfs.mainnet.iotex.io",
+		LogLevel:                 int(slog.LevelDebug),
+		SequencerPubKey:          "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
 	}
 
 	defaultDebugConfig = &Config{
-		Risc0ServerEndpoint:    "localhost:4001",
-		Halo2ServerEndpoint:    "localhost:4002",
-		ZKWasmServerEndpoint:   "localhost:4003",
-		WasmServerEndpoint:     "localhost:4004",
-		ChainEndpoint:          "https://babel-api.testnet.iotex.io",
-		ProjectContractAddress: "0x2339644f65c371Ca36b335A7eC3EB8AD8CBd5F51",
-		DatabaseDSN:            "postgres://test_user:test_passwd@localhost:5432/test?sslmode=disable",
-		BootNodeMultiAddr:      "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
-		ProverContractAddress:  "0x2B9BC8eC74E7F2526045eb13fFa37b10e0d40464",
-		ProverPrivateKey:       "dbfe03b0406549232b8dccc04be8224fcc0afa300a33d4f335dcfdfead861c85",
-		IoTeXChainID:           2,
-		SchedulerEpoch:         20,
-		IPFSEndpoint:           "ipfs.mainnet.iotex.io",
-		ProjectFileDirectory:   "../../test/project",
-		ProjectCacheDirectory:  "./project_cache",
-		LogLevel:               int(slog.LevelDebug),
-		SequencerPubKey:        "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
+		Risc0ServerEndpoint:      "localhost:4001",
+		Halo2ServerEndpoint:      "localhost:4002",
+		ZKWasmServerEndpoint:     "localhost:4003",
+		WasmServerEndpoint:       "localhost:4004",
+		ChainEndpoint:            "https://babel-api.testnet.iotex.io",
+		ProjectContractAddress:   "0x57872cF4b6D02053D5486F6882062c41a0F5C26a",
+		DatabaseDSN:              "postgres://test_user:test_passwd@localhost:5432/test?sslmode=disable",
+		BootNodeMultiAddr:        "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
+		ProverContractAddress:    "0x5B1AE92ed4a819C0B0e957a67F8dc39c0aC17Dfc",
+		ProverOperatorPrivateKey: "dbfe03b0406549232b8dccc04be8224fcc0afa300a33d4f335dcfdfead861c85",
+		IoTeXChainID:             2,
+		SchedulerEpoch:           20,
+		IPFSEndpoint:             "ipfs.mainnet.iotex.io",
+		ProjectCacheDirectory:    "./project_cache",
+		LogLevel:                 int(slog.LevelDebug),
+		SequencerPubKey:          "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
 	}
 
 	defaultTestConfig = &Config{
-		Risc0ServerEndpoint:    "localhost:14001",
-		Halo2ServerEndpoint:    "localhost:14002",
-		ZKWasmServerEndpoint:   "localhost:14003",
-		WasmServerEndpoint:     "localhost:14004",
-		ChainEndpoint:          "https://babel-api.testnet.iotex.io",
-		ProjectContractAddress: "",
-		DatabaseDSN:            "postgres://test_user:test_passwd@localhost:15432/test?sslmode=disable",
-		BootNodeMultiAddr:      "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
-		ProverContractAddress:  "",
-		ProverPrivateKey:       "",
-		IoTeXChainID:           2,
-		SchedulerEpoch:         20,
-		IPFSEndpoint:           "ipfs.mainnet.iotex.io",
-		ProjectFileDirectory:   "./testdata",
-		LogLevel:               int(slog.LevelDebug),
-		SequencerPubKey:        "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
+		Risc0ServerEndpoint:      "localhost:14001",
+		Halo2ServerEndpoint:      "localhost:14002",
+		ZKWasmServerEndpoint:     "localhost:14003",
+		WasmServerEndpoint:       "localhost:14004",
+		ChainEndpoint:            "https://babel-api.testnet.iotex.io",
+		ProjectContractAddress:   "",
+		DatabaseDSN:              "postgres://test_user:test_passwd@localhost:15432/test?sslmode=disable",
+		BootNodeMultiAddr:        "/dns4/bootnode-0.testnet.iotex.one/tcp/4689/ipfs/12D3KooWFnaTYuLo8Mkbm3wzaWHtUuaxBRe24Uiopu15Wr5EhD3o",
+		ProverContractAddress:    "",
+		ProverOperatorPrivateKey: "",
+		IoTeXChainID:             2,
+		SchedulerEpoch:           20,
+		IPFSEndpoint:             "ipfs.mainnet.iotex.io",
+		ProjectFileDirectory:     "./testdata",
+		LogLevel:                 int(slog.LevelDebug),
+		SequencerPubKey:          "0x04df6acbc5b355aabfb2145b36b20b7942c831c245c423a20b189fab4cf3a3dba3d564080841f2eb4890c118ca5e0b80b25f81269621c5e28273a962996c109afa",
 	}
 )
 

cmd/prover/config/config_test.go

@@ -16,20 +16,20 @@ func TestConfig_Init(t *testing.T) {
 	t.Run("UseEnvConfig", func(t *testing.T) {
 		os.Clearenv()
 		expected := config.Config{
-			Risc0ServerEndpoint:    "risc0:1111",
-			Halo2ServerEndpoint:    "halo2:2222",
-			ZKWasmServerEndpoint:   "zkwasm:3333",
-			WasmServerEndpoint:     "wasm:4444",
-			ChainEndpoint:          "http://abc.def.com",
-			ProjectContractAddress: "0x123",
-			DatabaseDSN:            "postgres://root@localhost/abc?ext=666",
-			BootNodeMultiAddr:      "/dsn4/abc/123",
-			ProverContractAddress:  "0x456",
-			IoTeXChainID:           5,
-			SchedulerEpoch:         720,
-			IPFSEndpoint:           "abc.ipfs.net",
-			ProverPrivateKey:       "private key",
-			ProjectFileDirectory:   "/path/to/project/configs",
+			Risc0ServerEndpoint:      "risc0:1111",
+			Halo2ServerEndpoint:      "halo2:2222",
+			ZKWasmServerEndpoint:     "zkwasm:3333",
+			WasmServerEndpoint:       "wasm:4444",
+			ChainEndpoint:            "http://abc.def.com",
+			ProjectContractAddress:   "0x123",
+			DatabaseDSN:              "postgres://root@localhost/abc?ext=666",
+			BootNodeMultiAddr:        "/dsn4/abc/123",
+			ProverContractAddress:    "0x456",
+			IoTeXChainID:             5,
+			SchedulerEpoch:           720,
+			IPFSEndpoint:             "abc.ipfs.net",
+			ProverOperatorPrivateKey: "private key",
+			ProjectFileDirectory:     "/path/to/project/configs",
 		}
 
 		_ = os.Setenv("RISC0_SERVER_ENDPOINT", expected.Risc0ServerEndpoint)
@@ -44,7 +44,7 @@ func TestConfig_Init(t *testing.T) {
 		_ = os.Setenv("SCHEDULER_EPOCH", strconv.FormatUint(expected.SchedulerEpoch, 10))
 		_ = os.Setenv("PROJECT_CONTRACT_ADDRESS", expected.ProjectContractAddress)
 		_ = os.Setenv("IPFS_ENDPOINT", expected.IPFSEndpoint)
-		_ = os.Setenv("PROVER_PRIVATE_KEY", expected.ProverPrivateKey)
+		_ = os.Setenv("PROVER_OPERATOR_PRIVATE_KEY", expected.ProverOperatorPrivateKey)
 		_ = os.Setenv("PROJECT_FILE_DIRECTORY", expected.ProjectFileDirectory)
 
 		c := &config.Config{}

cmd/prover/main.go

@@ -7,14 +7,17 @@ import (
 	"os/signal"
 	"syscall"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/pkg/errors"
 
 	"github.com/machinefi/sprout/cmd/prover/config"
 	"github.com/machinefi/sprout/p2p"
 	"github.com/machinefi/sprout/project"
 	"github.com/machinefi/sprout/scheduler"
+	"github.com/machinefi/sprout/smartcontracts/go/prover"
 	"github.com/machinefi/sprout/task"
 	"github.com/machinefi/sprout/vm"
 )
@@ -46,28 +49,42 @@ func main() {
 		log.Fatal(err)
 	}
 
-	sk, err := crypto.HexToECDSA(conf.ProverPrivateKey)
+	sk, err := crypto.HexToECDSA(conf.ProverOperatorPrivateKey)
 	if err != nil {
 		log.Fatal(errors.Wrap(err, "failed to parse prover private key"))
 	}
-	proverID := crypto.PubkeyToAddress(sk.PublicKey).String()
-
-	slog.Info("my prover id", "prover_id", proverID)
+	proverOperatorAddress := crypto.PubkeyToAddress(sk.PublicKey)
 
 	sequencerPubKey, err := hexutil.Decode(conf.SequencerPubKey)
 	if err != nil {
 		log.Fatal(errors.Wrap(err, "failed to decode sequencer pubkey"))
 	}
 
-	taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, proverID)
+	// TODO move to utils contract
+	client, err := ethclient.Dial(conf.ChainEndpoint)
+	if err != nil {
+		log.Fatal(err, "failed to dial chain endpoint")
+	}
+	instance, err := prover.NewProver(common.HexToAddress(conf.ProverContractAddress), client)
+	if err != nil {
+		log.Fatal(errors.Wrap(err, "failed to new prover contract instance"))
+	}
+	proverID, _, err := instance.OwnerOfOperator(nil, proverOperatorAddress)
+	if err != nil {
+		log.Fatal(errors.Wrap(err, "failed to query operator's prover id"))
+	}
+
+	slog.Info("my prover id", "prover_id", proverID.Uint64())
+
+	taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, proverID.Uint64())
 
 	pubSubs, err := p2p.NewPubSubs(taskProcessor.HandleP2PData, conf.BootNodeMultiAddr, conf.IoTeXChainID)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	if err := scheduler.Run(conf.SchedulerEpoch, conf.ChainEndpoint, conf.ProverContractAddress,
-		conf.ProjectContractAddress, conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers,
+		conf.ProjectContractAddress, conf.ProjectFileDirectory, proverID.Uint64(), pubSubs, taskProcessor.HandleProjectProvers,
 		projectManager.GetCachedProjectIDs); err != nil {
 		log.Fatal(err)
 	}

cmd/tests/init.go

@@ -120,26 +120,25 @@ func runProver(conf *proverconfig.Config) {
 		log.Fatal(err)
 	}
 
-	sk, err := crypto.HexToECDSA(conf.ProverPrivateKey)
+	sk, err := crypto.HexToECDSA(conf.ProverOperatorPrivateKey)
 	if err != nil {
 		log.Fatal(errors.Wrap(err, "failed parse prover private key"))
 	}
-	proverID := crypto.PubkeyToAddress(sk.PublicKey).String()
 
 	sequencerPubKey, err := hexutil.Decode(conf.SequencerPubKey)
 	if err != nil {
 		log.Fatal(errors.Wrap(err, "failed to decode sequencer pubkey"))
 	}
 
-	taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, proverID)
+	taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, 1)
 
 	pubSubs, err := p2p.NewPubSubs(taskProcessor.HandleP2PData, conf.BootNodeMultiAddr, conf.IoTeXChainID)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	if err := scheduler.Run(conf.SchedulerEpoch, conf.ChainEndpoint, conf.ProverContractAddress, conf.ProjectContractAddress,
-		conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers, projectManager.GetCachedProjectIDs); err != nil {
+		conf.ProjectFileDirectory, 1, pubSubs, taskProcessor.HandleProjectProvers, projectManager.GetCachedProjectIDs); err != nil {
 		log.Fatal(err)
 	}