update workerid

server/service.go

@@ -6,6 +6,7 @@ import (
 	"github.com/odysseus/nodemanager/utils"
 	omanager "github.com/odysseus/odysseus-protocol/gen/proto/go/nodemanager/v1"
 	log "github.com/sirupsen/logrus"
+	"strconv"
 	"strings"
 )
 
@@ -63,6 +64,14 @@ func (n *NodeManagerService) DispatchTask(ctx context.Context, request *omanager
 	if worker.online == false {
 		return nil, errors.New("worker offline")
 	}
+	{
+		nonceds := strings.Split(mids[1], ":")
+		nonce, _ := strconv.ParseInt(nonceds[0], 10, 64)
+		if nonce < int64(worker.nonce) {
+			return nil, errors.New("expired worker nonce")
+		}
+	}
+
 	dtask := &dispatchTask{
 		task:  request.TaskData,
 		errCh: make(chan error, 1),

server/workerstatu.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/gomodule/redigo/redis"
+	"github.com/google/uuid"
 	"github.com/odysseus/nodemanager/config"
 	log "github.com/sirupsen/logrus"
 	"strconv"
@@ -67,7 +68,7 @@ func (wm *WorkerManager) AddWorkerFirst(worker *Worker) error {
 		_ = device // todo: set priority with device info.
 		for m := 0; m < config.GetConfig().GetWorkerMultiple(); m++ {
 			// add worker to redis queue
-			if err := wm.rdb.RPush(context.Background(), config.WORKER_QUEUE_PREFIX+strconv.Itoa(priority), workerId(worker)).Err(); err != nil {
+			if err := wm.rdb.RPush(context.Background(), config.WORKER_QUEUE_PREFIX+strconv.Itoa(priority), workerUid(worker)).Err(); err != nil {
 				continue
 			}
 		}
@@ -115,7 +116,7 @@ func (wm *WorkerManager) AddWorkerSingle(worker *Worker) error {
 	{
 		// add worker to redis queue
 		priority := 0
-		if err := wm.rdb.RPush(context.Background(), config.WORKER_QUEUE_PREFIX+strconv.Itoa(priority), workerId(worker)).Err(); err != nil {
+		if err := wm.rdb.RPush(context.Background(), config.WORKER_QUEUE_PREFIX+strconv.Itoa(priority), workerUid(worker)).Err(); err != nil {
 			log.WithError(err).Error("add worker back to queue failed.")
 		} else {
 			log.WithField("worker", worker.workerAddr).Info("add worker back to queue success.")
@@ -269,6 +270,12 @@ func workerLastTaskTmKey(w *Worker) string {
 	return config.WORKER_LAST_TASK_TM_PREFIX + w.workerAddr
 }
 
+func workerUid(w *Worker) string {
+	id := workerId(w)
+	uuid := uuid.NewString()
+	return fmt.Sprintf("%s:%s", id, uuid[:4])
+}
+
 func workerId(w *Worker) string {
 	return fmt.Sprintf("%s_%d", w.workerAddr, w.nonce)
 }