[impl/memory] Enforce named tasks during txn.

impl/memory/taskqueue.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ skipping 46 matching lines @@
 func (t *taskqueueImpl) AddMulti(tasks []*tq.Task, queueName string, cb tq.RawTaskCB) error {
         t.Lock()
         defer t.Unlock()
 
         q, err := t.getQueueLocked(queueName)
         if err != nil {
                 return err
         }
 
         for _, task := range tasks {
-                task, err := prepTask(t.ctx, task, q, t.ns)
+                task, err := prepTask(t.ctx, task, q, t.ns, false)
                 if err == nil {
                         err = q.addTask(task)
                 }
                 if err != nil {
                         cb(nil, err)
                 } else {
                         cb(task.Duplicate(), nil)
                 }
         }
         return nil
@@ skipping 84 matching lines @@
 type taskqueueTxnImpl struct {
         *txnTaskQueueData
 
         ctx context.Context
         ns  string
 }
 
 var _ tq.RawInterface = (*taskqueueTxnImpl)(nil)
 
 func (t *taskqueueTxnImpl) addLocked(task *tq.Task, q *sortedQueue) (*tq.Task, error) {
-        toSched, err := prepTask(t.ctx, task, q, t.ns)
+        toSched, err := prepTask(t.ctx, task, q, t.ns, true)
         if err != nil {
                 return nil, err
         }
 
         numTasks := 0
         for _, vs := range t.anony {
                 numTasks += len(vs)
         }
         if numTasks+1 > 5 {
                 // transactional tasks are actually implemented 'for real' as Actions which
@@ skipping 73 matching lines @@
         t.Lock()
         defer t.Unlock()
         t.setConstraintsLocked(*c)
         return nil
 }
 
 func (t *taskqueueTxnImpl) GetTestable() tq.Testable { return &taskQueueTestable{t.ns, t} }
 
 ////////////////////////// private functions ///////////////////////////////////
 
-func prepTask(c context.Context, task *tq.Task, q *sortedQueue, ns string) (*tq.Task, error) {
+func prepTask(c context.Context, task *tq.Task, q *sortedQueue, ns string, inTxn bool) (*tq.Task, error) {
         toSched := task.Duplicate()
 
+        if inTxn && task.Name != "" {
+                return nil, fmt.Errorf("taskqueue: INVALID_TASK_NAME: cannot add named task %q in transaction", task.Name)
+        }
+
         if toSched.ETA.IsZero() {
                 toSched.ETA = clock.Now(c).Add(toSched.Delay)
         } else if toSched.Delay != 0 {
                 panic("taskqueue: both Delay and ETA are set")
         }
         toSched.Delay = 0
 
         switch toSched.Method {
         // Methods that can have payloads.
         case "":
@@ skipping 34 matching lines @@
         } else {
                 if !validTaskName.MatchString(toSched.Name) {
                         return nil, errInvalidTaskName
                 }
         }
 
         return toSched, nil
 }
 
 func taskNamespace(task *tq.Task) string { return task.Header.Get(currentNamespace) }