disallow manual trigger of completed tasks

core/taskengine/engine.go

@@ -566,6 +566,10 @@ func (n *Engine) TriggerTask(user *model.User, payload *avsproto.UserTriggerTask
 		return nil, err
 	}
 
+	if !task.IsRunable() {
+		return nil, grpcstatus.Errorf(codes.FailedPrecondition, TaskIsNotRunable)
+	}
+
 	if !task.OwnedBy(user.Address) {
 		// only the owner of a task can trigger it
 		return nil, grpcstatus.Errorf(codes.NotFound, TaskNotFoundError)

core/taskengine/engine_test.go

@@ -331,7 +331,6 @@ func TestTriggerSync(t *testing.T) {
 		t.Errorf("invalid triggered block. expect 101 got %d", execution.TriggerMetadata.BlockNumber)
 	}
 }
-
 func TestTriggerAsync(t *testing.T) {
 	db := testutil.TestMustDB()
 	defer storage.Destroy(db.(*storage.BadgerStorage))
@@ -413,3 +412,46 @@ func TestTriggerAsync(t *testing.T) {
 		t.Errorf("invalid execution status, expected completed but got %s", avsproto.TaskStatus_name[int32(executionStatus.Status)])
 	}
 }
+
+func TestTriggerCompletedTaskReturnError(t *testing.T) {
+	db := testutil.TestMustDB()
+	defer storage.Destroy(db.(*storage.BadgerStorage))
+
+	config := testutil.GetAggregatorConfig()
+	n := New(db, config, nil, testutil.GetLogger())
+
+	// Now create a test task
+	tr1 := testutil.RestTask()
+	tr1.Name = "t1"
+	tr1.MaxExecution = 1
+	// salt 0
+	tr1.SmartWalletAddress = "0x7c3a76086588230c7B3f4839A4c1F5BBafcd57C6"
+	result, _ := n.CreateTask(testutil.TestUser1(), tr1)
+
+	resultTrigger, err := n.TriggerTask(testutil.TestUser1(), &avsproto.UserTriggerTaskReq{
+		TaskId: result.Id,
+		TriggerMetadata: &avsproto.TriggerMetadata{
+			BlockNumber: 101,
+		},
+		IsBlocking: true,
+	})
+
+	if err != nil || resultTrigger == nil {
+		t.Errorf("expected trigger succesfully but got error: %s", err)
+	}
+
+	fmt.Println(resultTrigger)
+	// Now the task has reach its max run, and canot run anymore
+	resultTrigger, err = n.TriggerTask(testutil.TestUser1(), &avsproto.UserTriggerTaskReq{
+		TaskId: result.Id,
+		TriggerMetadata: &avsproto.TriggerMetadata{
+			BlockNumber: 101,
+		},
+		IsBlocking: true,
+	})
+
+	if err == nil || resultTrigger != nil {
+		t.Errorf("expect trigger error but succeed")
+	}
+
+}

core/taskengine/errors.go

@@ -20,6 +20,7 @@ const (
 
 	TaskStorageCorruptedError = "task data storage is corrupted"
 	TaskIDMissing             = "Missing task id in request"
+	TaskIsNotRunable          = "The workflow is not in a runable status, it has reached the limit execution or the expiration time"
 
 	InvalidCursor          = "cursor is not valid"
 	InvalidPaginationParam = "item per page is not valid"

model/task.go

@@ -130,6 +130,18 @@ func (t *Task) OwnedBy(address common.Address) bool {
 	return strings.EqualFold(t.Owner, address.Hex())
 }
 
+// A task is runable when both of these condition are matched
+//  1. Its max execution has not reached
+//  2. Its expiration time has not reached
+func (t *Task) IsRunable() bool {
+	// When MaxExecution is 0, it is unlimited run
+	reachedMaxRun := t.MaxExecution > 0 && t.TotalExecution >= t.MaxExecution
+
+	reachedExpiredTime := t.ExpiredAt > 0 && time.Unix(t.ExpiredAt, 0).Before(time.Now())
+
+	return !reachedMaxRun && !reachedExpiredTime
+}
+
 // Given a task key generated from Key(), extract the ID part
 func TaskKeyToId(key []byte) []byte {
 	// <43-byte>:<43-byte>: