update config

config/Config.go

@@ -17,12 +17,21 @@ type KafkaConfig struct {
 	TaskTopic    string `json:"task_topic" toml:"task_topic"`
 }
 
+type MysqlConfig struct {
+	Host   string `json:"host" toml:"host"`
+	Port   int    `json:"port" toml:"port"`
+	User   string `json:"user" toml:"user"`
+	Passwd string `json:"password" toml:"password"`
+	DbName string `json:"database" toml:"database"`
+}
+
 type Config struct {
 	Endpoint   string      `json:"endpoint" toml:"endpoint"`
 	MetricPort int         `json:"metrics_port" toml:"metrics_port"`
 	Routines   int         `json:"routines" toml:"routines"`
 	Redis      RedisConfig `json:"redis" toml:"redis"`
 	Kafka      KafkaConfig `json:"kafka" toml:"kafka"`
+	DbConfig   MysqlConfig `json:"mysql" toml:"mysql"`
 }
 
 var _cfg *Config = nil

go.mod

@@ -8,9 +8,10 @@ require (
 	github.com/astaxie/beego v1.12.3
 	github.com/ethereum/go-ethereum v1.13.10
 	github.com/gogo/protobuf v1.3.2
-	github.com/google/uuid v1.4.0
+	github.com/google/uuid v1.5.0
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible
 	github.com/odysseus/odysseus-protocol v0.0.0-00010101000000-000000000000
+	github.com/odysseus/payment v0.0.0-00010101000000-000000000000
 	github.com/prometheus/client_golang v1.18.0
 	github.com/redis/go-redis/v9 v9.4.0
 	github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5
@@ -31,11 +32,14 @@ require (
 	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/go-sql-driver/mysql v1.7.1 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
+	github.com/gomodule/redigo v2.0.0+incompatible // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/holiman/uint256 v1.2.4 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
@@ -79,3 +83,5 @@ require (
 )
 
 replace github.com/odysseus/odysseus-protocol => ../odysseus-protocol
+
+replace github.com/odysseus/payment => ../payment

go.sum

@@ -67,6 +67,8 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-redis/redis v6.14.2+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
+github.com/go-sql-driver/mysql v1.7.1 h1:lUIinVbN1DY0xBg0eMOzmmtGoHwWBbvnWubQUrtU8EI=
+github.com/go-sql-driver/mysql v1.7.1/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
@@ -87,6 +89,7 @@ github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8l
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb h1:PBC98N2aIaM3XXiurYmW7fx4GZkL8feAMVq7nEjURHk=
 github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
 github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -94,8 +97,8 @@ github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
-github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
-github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
+github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
 github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
@@ -105,6 +108,7 @@ github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9
 github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
 github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
@@ -148,9 +152,11 @@ github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible h1:Y6sqxHMyB1D2YSzWkL
 github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible/go.mod h1:ZQnN8lSECaebrkQytbHj4xNgtg8CR7RYXnPok8e0EHA=
 github.com/lestrrat-go/strftime v1.0.6 h1:CFGsDEt1pOpFNU+TJB0nhz9jl+K0hZSLE205AhTIGQQ=
 github.com/lestrrat-go/strftime v1.0.6/go.mod h1:f7jQKgV5nnJpYgdEasS+/y7EsTb8ykN2z68n3TtcTaw=
+github.com/lib/pq v1.0.0 h1:X5PMW56eZitiTeO7tKzZxFCSpbFZJtkMMooicw2us9A=
 github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
 github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
+github.com/mattn/go-sqlite3 v2.0.3+incompatible h1:gXHsfypPkaMZrKbD5209QV9jbUTJKjyR5WD3HYQSd+U=
 github.com/mattn/go-sqlite3 v2.0.3+incompatible/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 h1:jWpvCLoY8Z/e3VKvlsiIGKtc+UG6U5vzxaoagmhXfyg=

server/node.go

@@ -6,10 +6,13 @@ import (
 	"github.com/IBM/sarama"
 	"github.com/gogo/protobuf/proto"
 	odysseus "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
+	"github.com/odysseus/payment/cachedata"
+	"github.com/odysseus/payment/model"
 	"github.com/odysseus/scheduler/config"
 	"github.com/odysseus/scheduler/utils"
 	"github.com/redis/go-redis/v9"
 	log "github.com/sirupsen/logrus"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -20,6 +23,7 @@ type Node struct {
 	quit          chan struct{}
 	conf          *config.Config
 	kafkaProducer sarama.AsyncProducer
+	cache         *cachedata.CacheData
 	wg            sync.WaitGroup
 }
 
@@ -30,10 +34,23 @@ func NewNode() *Node {
 		Password: redisConfig.Password,
 		DbIndex:  redisConfig.DbIndex,
 	})
+	dbconf := config.GetConfig().DbConfig
+	pay := cachedata.NewCacheData(context.Background(), cachedata.RedisConnParam{
+		Addr:     redisConfig.Addr,
+		Password: redisConfig.Password,
+		DbIndex:  redisConfig.DbIndex,
+	}, model.DbConfig{
+		Host:   dbconf.Host,
+		Port:   dbconf.Port,
+		User:   dbconf.User,
+		Passwd: dbconf.Passwd,
+		DbName: dbconf.DbName,
+	})
 	node := &Node{
-		rdb:  rdb,
-		quit: make(chan struct{}),
-		conf: config.GetConfig(),
+		cache: pay,
+		rdb:   rdb,
+		quit:  make(chan struct{}),
+		conf:  config.GetConfig(),
 	}
 	brokers := strings.Split(node.conf.Kafka.Brokers, ";")
 	node.kafkaProducer, _ = utils.NewKafkaProducer(brokers)
@@ -78,6 +95,10 @@ func (n *Node) Loop(idx int) {
 		} else {
 			log.WithField("taskid", task.TaskId).Debug("post task result")
 		}
+		uid, _ := strconv.ParseInt(task.TaskUid, 10, 64)
+		fee, _ := strconv.ParseInt(task.TaskFee, 10, 64)
+		n.cache.RollbackForFee(uid, fee)
+
 		return err
 	}
 	postReceipt := func(task *odysseus.TaskContent, result *odysseus.TaskResponse, err error) error {
@@ -117,11 +138,12 @@ func (n *Node) Loop(idx int) {
 				worker, err := PopWorker(n.rdb)
 				if err == ErrNoWorker {
 					result := &odysseus.TaskResponse{
-						TaskUuid:      task.TaskUuid,
-						TaskUid:       task.TaskUid,
-						TaskFee:       task.TaskFee,
-						TaskIsSucceed: false,
-						TaskError:     err.Error(),
+						TaskUuid:            task.TaskUuid,
+						TaskUid:             task.TaskUid,
+						TaskFee:             task.TaskFee,
+						TaskIsSucceed:       false,
+						TaskError:           err.Error(),
+						TaskExecuteDuration: 0,
 					}
 					postReceipt(task, result, err)
 					err = postResult(task, result)