update protocol

cmd/demoworker/main.go

@@ -141,7 +141,7 @@ func main() {
 			msg := &omanager.WorkerMessage{
 				Message: &omanager.WorkerMessage_SubmitTaskResult{
 					SubmitTaskResult: &omanager.SubmitTaskResult{
-						TaskUuid:           b.PushTaskMessage.TaskUuid,
+						TaskId:             b.PushTaskMessage.TaskId,
 						ContainerSignature: make([]byte, 65),
 						MinerSignature:     make([]byte, 65),
 						TaskResultBody:     []byte(demoResult),

config.toml

@@ -3,7 +3,6 @@ local_host="127.0.0.1"
 port=10001
 metrics_port = 28010
 private_key = "E671C143A110C239B563F702E9F4017CA6B2B2912F675EED9AA4FED684EB30CC"
-enable_pay = false
 
 [redis]
 addr="127.0.0.1:6379"
@@ -22,6 +21,7 @@ brokers="127.0.0.1:9092"
 receipt_topic="taskreceipt"
 
 [ticker]
+#second
 heart_beat = 10
 status_ticker = 10
 device_info_ticker = 120

server/service.go

@@ -33,6 +33,7 @@ func (n *NodeManagerService) ManagerList(ctx context.Context, request *omanager.
 
 func (n *NodeManagerService) RegisterWorker(client omanager.NodeManagerService_RegisterWorkerServer) error {
 	uuid := utils.GetSnowflakeId()
+
 	worker, err := n.node.wm.AddNewWorker(uuid, client)
 	if err != nil {
 		return err
@@ -65,7 +66,7 @@ func (n *NodeManagerService) DispatchTask(ctx context.Context, request *omanager
 	}
 
 	res := new(omanager.DispatchTaskResponse)
-	res.TaskUuid = request.TaskData.TaskUuid
+	res.TaskId = request.TaskData.TaskId
 	res.Miner = request.Miner
 	return res, nil
 }

server/workerManager.go

@@ -157,13 +157,13 @@ func (wm *WorkerManager) doCallback(hook string, response *odysseus.TaskResponse
 	if err != nil {
 		log.WithError(err).Error("marshal task response failed")
 	} else {
-		log.WithField("taskid", response.TaskUuid).Debug("marshal task response")
+		log.WithField("task-id", response.TaskId).Debug("marshal task response")
 	}
 	err = utils.Post(hook, d)
 	if err != nil {
 		log.WithError(err).Error("post task result failed")
 	} else {
-		log.WithField("taskid", response.TaskUuid).Debug("post task result")
+		log.WithField("task-id", response.TaskId).Debug("post task result")
 	}
 }
 
@@ -280,8 +280,9 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 			task := dtask.task
 			taskMsg := new(omanager.ManagerMessage_PushTaskMessage)
 			taskMsg.PushTaskMessage = &omanager.PushTaskMessage{
-				TaskUuid:  task.TaskUuid,
+				TaskId:    task.TaskId,
 				TaskType:  task.TaskType,
+				TaskKind:  task.TaskKind,
 				Workload:  uint64(task.TaskWorkload),
 				TaskCmd:   task.TaskCmd,
 				TaskParam: task.TaskParam,
@@ -290,7 +291,7 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 			callback = func(err error) bool {
 				if err == nil {
 					// add task to cache.
-					worker.recentTask.Add(task.TaskUuid, task)
+					worker.recentTask.Add(task.TaskId, task)
 				}
 				log.WithField("worker", worker.uuid).Info("dispatch task to worker")
 
@@ -303,23 +304,23 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 			}
 		case result := <-worker.resultCh:
 			// verify result and make a new signature.
-			data, exist := worker.recentTask.Get(result.TaskUuid)
+			data, exist := worker.recentTask.Get(result.TaskId)
 			if !exist {
 				log.WithField("worker", worker.uuid).Error("task not found for verify result")
 				continue
 			}
 
 			task := data.(*odysseus.TaskContent)
-			if result.TaskUuid != task.TaskUuid {
+			if result.TaskId != task.TaskId {
 				log.WithField("worker", worker.uuid).Error("task id not match")
 				continue
 			}
 
-			log.WithField("task-uuid", task.TaskUuid).WithField("result", result).Debug("got task result")
+			log.WithField("task-id", task.TaskId).WithField("result", result).Debug("got task result")
 
 			if result.IsSuccessed == false {
 				taskResponse := &odysseus.TaskResponse{
-					TaskUuid:         task.TaskUuid,
+					TaskId:           task.TaskId,
 					TaskResultHeader: result.TaskResultHeader,
 					TaskResultBody:   result.TaskResultBody,
 					TaskUid:          task.TaskUid,
@@ -350,7 +351,7 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 				// miner_signature = sign(hash((task_id+hash(task_param)+hash(task_result)))
 				paramHash := crypto.Keccak256Hash(task.TaskParam)
 				resultHash := crypto.Keccak256Hash(result.TaskResultBody)
-				dataHash := crypto.Keccak256Hash(utils.CombineBytes([]byte(result.TaskUuid), paramHash[:], resultHash[:]))
+				dataHash := crypto.Keccak256Hash(utils.CombineBytes([]byte(result.TaskId), paramHash[:], resultHash[:]))
 				minerPubkey, _ := utils.HexToPubkey(worker.publicKey) // todo: get miner pubkey
 				verified := ecdsa.VerifyASN1(minerPubkey, dataHash[:], result.MinerSignature)
 				log.WithField("minerSignatureVerify", verified).Debug("miner signature verify")
@@ -365,7 +366,7 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 			//manager_signature = sign(hash((task_id+hash(task_param)+hash(task_result)+container_signature+miner_signature+workload))
 			paramHash := crypto.Keccak256Hash(task.TaskParam)
 			resultHash := crypto.Keccak256Hash(result.TaskResultBody)
-			dataHash := crypto.Keccak256Hash(utils.CombineBytes([]byte(result.TaskUuid), paramHash[:], resultHash[:],
+			dataHash := crypto.Keccak256Hash(utils.CombineBytes([]byte(result.TaskId), paramHash[:], resultHash[:],
 				worker.ProfitAccount().Bytes(), worker.WorkerAccount().Bytes(), result.ContainerSignature, result.MinerSignature, big.NewInt(int64(task.TaskWorkload)).Bytes()))
 
 			signature, err := wm.node.Sign(dataHash[:])
@@ -376,20 +377,24 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 
 			proof := new(omanager.ManagerMessage_ProofTaskResult)
 			proof.ProofTaskResult = &omanager.ProofTaskResult{
-				TaskUuid:         result.TaskUuid,
+				TaskId:           result.TaskId,
 				ManagerSignature: signature,
 				Workload:         uint64(task.TaskWorkload),
 				ContainerPubkey:  utils.CombineBytes(task.ContainerPubkey),
 			}
+			log.WithFields(log.Fields{
+				"task-id":  result.TaskId,
+				"workload": task.TaskWorkload,
+			}).Debug("send proof to worker")
 			msg.Message = proof
 
 			callback = func(err error) bool {
 				if err == nil {
 					// remove task from cache.
-					worker.recentTask.Remove(result.TaskUuid)
+					worker.recentTask.Remove(result.TaskId)
 				}
 				taskResponse := &odysseus.TaskResponse{
-					TaskUuid:         task.TaskUuid,
+					TaskId:           task.TaskId,
 					TaskResultHeader: result.TaskResultHeader,
 					TaskResultBody:   result.TaskResultBody,
 					TaskUid:          task.TaskUid,
@@ -535,9 +540,9 @@ func (wm *WorkerManager) Payment(task *odysseus.TaskContent) error {
 func (wm *WorkerManager) makeReceipt(worker *Worker, task *odysseus.TaskContent, result *omanager.SubmitTaskResult, err error) *odysseus.TaskReceipt {
 	now := uint64(time.Now().UnixNano())
 	receipt := &odysseus.TaskReceipt{
-		TaskUuid:            task.TaskUuid,
-		TaskTimestamp:       task.TaskTimestamp,
 		TaskId:              task.TaskId,
+		TaskTimestamp:       task.TaskTimestamp,
+		TaskType:            task.TaskType,
 		TaskUid:             task.TaskUid,
 		TaskWorkload:        task.TaskWorkload,
 		TaskDuration:        int64(now-task.TaskTimestamp) / 1000,