diff --git a/appengine/tq/tq.go b/appengine/tq/tq.go
index 6f974dd..397c465 100644
--- a/appengine/tq/tq.go
+++ b/appengine/tq/tq.go
@@ -59,12 +59,28 @@
 	// Tasks are routed based on type of the payload message, see RegisterTask.
 	Payload proto.Message
 
+	// NamePrefix, if not empty, is a string that will be prefixed to the task's
+	// name. Characters in NamePrefix must be appropriate task queue name
+	// characters. NamePrefix can be useful because the Task Queue system allows
+	// users to search for tasks by prefix.
+	//
+	// Lexicographically close names can cause hot spots in the Task Queues
+	// backend. If NamePrefix is specified, users should try and ensure that
+	// it is friendly to sharding (e.g., begins with a hash string).
+	//
+	// Setting NamePrefix and/or DeduplicationKey will result in a named task
+	// being generated. This task can be cancelled using DeleteTask.
+	NamePrefix string
+
 	// DeduplicationKey is optional unique key of the task.
 	//
 	// If a task of a given proto type with a given key has already been enqueued
 	// recently, this task will be silently ignored.
 	//
 	// Such tasks can only be used outside of transactions.
+	//
+	// Setting NamePrefix and/or DeduplicationKey will result in a named task
+	// being generated. This task can be cancelled using DeleteTask.
 	DeduplicationKey string
 
 	// Title is optional string that identifies the task in HTTP logs.
@@ -93,6 +109,40 @@
 	RetryOptions *taskqueue.RetryOptions
 }
 
+// Name generates and returns the task's name.
+//
+// If the task is not a named task (doesn't have NamePrefix or DeduplicationKey
+// set), this will return an empty string.
+func (task *Task) Name() string {
+	if task.NamePrefix == "" && task.DeduplicationKey == "" {
+		return ""
+	}
+
+	parts := make([]string, 0, 2)
+
+	if task.NamePrefix != "" {
+		parts = append(parts, task.NamePrefix)
+	}
+
+	// There's some weird restrictions on what characters are allowed inside task
+	// names. Lexicographically close names also cause hot spot problems in the
+	// Task Queues backend. To avoid these two issues, we always use SHA256 hashes
+	// as task names. Also each task kind owns its own namespace of deduplication
+	// keys, so add task type to the digest as well.
+	if task.DeduplicationKey != "" {
+		h := sha256.New()
+		if task.Payload == nil {
+			panic("task must have a Payload")
+		}
+		h.Write([]byte(proto.MessageName(task.Payload)))
+		h.Write([]byte{0})
+		h.Write([]byte(task.DeduplicationKey))
+		parts = append(parts, hex.EncodeToString(h.Sum(nil)))
+	}
+
+	return strings.Join(parts, "-")
+}
+
 // Handler is called to handle one enqueued task.
 //
 // The passed context is produced by a middleware chain installed with
@@ -144,19 +194,22 @@
 	}
 }
 
-// AddTask submits given tasks to an appropriate task queue.
+// runBatchesPerQueue is a generic parallel task distributor. It solves the
+// problems that:
+//	- "tasks" may be assigned to different queues, and tasks assigned to the
+//    same queue should be batched together.
+//	- Any given batch may exceed queue operation limits, and thus needs to be
+//	  broken into multiple operations on sub-batches.
 //
-// It means, at some later time in some other GAE process, callbacks registered
-// as handlers for corresponding proto types will be called.
-//
-// If the given context is transactional, inherits the transaction. Note if
-// running outside of a transaction and multiple tasks are passed, the operation
-// is not atomic: it returns an error if at least one enqueue operation failed
-// (there's no way to figure out which one exactly).
-//
-// Returns only transient errors. Unlike regular Task Queue's Add,
-// ErrTaskAlreadyAdded is not considered an error.
-func (d *Dispatcher) AddTask(c context.Context, tasks ...*Task) error {
+// fn is called for each sub-batch assigned to each queue. All resulting errors
+// are then flattened. If no fn invocation returns any errors, nil will be
+// returned. If a single error is returned, this function will return that
+// error. If an errors.MultiError is returned, it and any embedded
+// MultiError (recursively) will be flattened into a single MultiError
+// containing only the non-nil errors. This simplifies user expectations.
+func (d *Dispatcher) runBatchesPerQueue(c context.Context, tasks []*Task,
+	fn func(c context.Context, queue string, tasks []*taskqueue.Task) error) error {
+
 	if len(tasks) == 0 {
 		return nil
 	}
@@ -167,10 +220,7 @@
 		if err != nil {
 			return err
 		}
-		if err := taskqueue.Add(c, queue, t); err != nil {
-			if err == taskqueue.ErrTaskAlreadyAdded {
-				return nil
-			}
+		if err := fn(c, queue, []*taskqueue.Task{t}); err != nil {
 			return transient.Tag.Apply(err)
 		}
 		return nil
@@ -187,42 +237,82 @@
 
 	// Enqueue in parallel, per-queue, split into batches based on Task Queue
 	// RPC limits (100 tasks per batch).
+	const maxBatchSize = 100
 	errs := make(chan error)
 	ops := 0
 	for q, tasks := range perQueue {
 		for len(tasks) > 0 {
-			count := 100
+			count := maxBatchSize
 			if count > len(tasks) {
 				count = len(tasks)
 			}
 			go func(q string, batch []*taskqueue.Task) {
-				errs <- taskqueue.Add(c, q, batch...)
+				errs <- fn(c, q, batch)
 			}(q, tasks[:count])
 			tasks = tasks[count:]
 			ops++
 		}
 	}
 
-	// Gather all errors throwing away ErrTaskAlreadyAdded.
-	var all errors.MultiError
+	all := errors.NewLazyMultiError(ops)
 	for i := 0; i < ops; i++ {
 		err := <-errs
-		if merr, yep := err.(errors.MultiError); yep {
-			for _, e := range merr {
-				if e != nil && e != taskqueue.ErrTaskAlreadyAdded {
-					all = append(all, e)
-				}
-			}
-		} else if err != nil && err != taskqueue.ErrTaskAlreadyAdded {
-			all = append(all, err)
+		if err != nil {
+			all.Assign(i, err)
 		}
 	}
 
-	if len(all) == 0 {
-		return nil
+	if err := flattenErrors(all.Get()); err != nil {
+		return transient.Tag.Apply(err)
 	}
+	return nil
+}
 
-	return transient.Tag.Apply(all)
+// AddTask submits given tasks to an appropriate task queue.
+//
+// It means, at some later time in some other GAE process, callbacks registered
+// as handlers for corresponding proto types will be called.
+//
+// If the given context is transactional or namespaced, inherits the
+// transaction/namespace. Note if running outside of a transaction and multiple
+// tasks are passed, the operation is not atomic: it returns an error if at
+// least one enqueue operation failed (there's no way to figure out which one
+// exactly).
+//
+// Returns only transient errors. Unlike regular Task Queue's Add,
+// ErrTaskAlreadyAdded is not considered an error.
+func (d *Dispatcher) AddTask(c context.Context, tasks ...*Task) error {
+	return d.runBatchesPerQueue(c, tasks, func(c context.Context, queue string, tasks []*taskqueue.Task) error {
+		if err := taskqueue.Add(c, queue, tasks...); err != nil {
+			return errors.Filter(err, taskqueue.ErrTaskAlreadyAdded)
+		}
+		return nil
+	})
+}
+
+// DeleteTask deletes the specified tasks from their queues.
+//
+// If the given context is transactional or namespaced, inherits the
+// transaction/namespace. Note if running outside of a transaction and multiple
+// tasks are passed, the operation is not atomic: it returns an error if at
+// least one enqueue operation failed (there's no way to figure out which one
+// exactly).
+//
+// Returns only transient errors. Unlike regular Task Queue's Delete,
+// attempts to delete an unknown or tombstoned task are not considered errors.
+func (d *Dispatcher) DeleteTask(c context.Context, tasks ...*Task) error {
+	return d.runBatchesPerQueue(c, tasks, func(c context.Context, queue string, tasks []*taskqueue.Task) error {
+		return errors.FilterFunc(taskqueue.Delete(c, queue, tasks...), func(err error) bool {
+			// Currently, the best way to detect an attempt to delete an unknown task
+			// is to check the string with tolerable error message phrases.
+			for _, phrase := range []string{"UNKNOWN_TASK", "TOMBSTONED_TASK"} {
+				if strings.Contains(err.Error(), phrase) {
+					return true
+				}
+			}
+			return false
+		})
+	})
 }
 
 // InstallRoutes installs appropriate HTTP routes in the router.
@@ -276,23 +366,9 @@
 		retryOpts = task.RetryOptions
 	}
 
-	// There's some weird restrictions on what characters are allowed inside task
-	// names. Lexicographically close names also cause hot spot problems in the
-	// Task Queues backend. To avoid these two issues, we always use SHA256 hashes
-	// as task names. Also each task kind owns its own namespace of deduplication
-	// keys, so add task type to the digest as well.
-	name := ""
-	if task.DeduplicationKey != "" {
-		h := sha256.New()
-		h.Write([]byte(handler.typeName))
-		h.Write([]byte{0})
-		h.Write([]byte(task.DeduplicationKey))
-		name = hex.EncodeToString(h.Sum(nil))
-	}
-
 	return &taskqueue.Task{
 		Path:         fmt.Sprintf("%s%s/%s", d.baseURL(), handler.queue, title),
-		Name:         name,
+		Name:         task.Name(),
 		Method:       "POST",
 		Payload:      blob,
 		ETA:          task.ETA,
@@ -414,3 +490,32 @@
 
 	return task, nil
 }
+
+////////////////////////////////////////////////////////////////////////////////
+
+// flattenErrors collapses a multi-dimensional MultiError space into a flat
+// MultiError, removing "nil" errors.
+//
+// If err is not an errors.MultiError, will return err directly.
+//
+// As a special case, if merr contains no non-nil errors, nil will be returned.
+func flattenErrors(err error) error {
+	var ret errors.MultiError
+	flattenErrorsRec(&ret, err)
+	if len(ret) == 0 {
+		return nil
+	}
+	return ret
+}
+
+func flattenErrorsRec(ret *errors.MultiError, err error) {
+	switch et := err.(type) {
+	case nil:
+	case errors.MultiError:
+		for _, e := range et {
+			flattenErrorsRec(ret, e)
+		}
+	default:
+		*ret = append(*ret, et)
+	}
+}
diff --git a/appengine/tq/tq_test.go b/appengine/tq/tq_test.go
index e3176d0..316be4d 100644
--- a/appengine/tq/tq_test.go
+++ b/appengine/tq/tq_test.go
@@ -57,6 +57,19 @@
 				next(c)
 			}))
 		}
+		runTasks := func(ctx context.Context) []int {
+			var codes []int
+			for _, tasks := range taskqueue.GetTestable(ctx).GetScheduledTasks() {
+				for _, task := range tasks {
+					// Execute the task.
+					req := httptest.NewRequest("POST", "http://example.com"+task.Path, bytes.NewReader(task.Payload))
+					rw := httptest.NewRecorder()
+					r.ServeHTTP(rw, req)
+					codes = append(codes, rw.Code)
+				}
+			}
+			return codes
+		}
 
 		Convey("Single task", func() {
 			var calls []proto.Message
@@ -74,6 +87,7 @@
 			err := d.AddTask(ctx, &Task{
 				Payload:          &duration.Duration{Seconds: 123},
 				DeduplicationKey: "abc",
+				NamePrefix:       "prefix",
 				Title:            "abc-def",
 				Delay:            30 * time.Second,
 			})
@@ -81,7 +95,7 @@
 
 			// Added the task.
 			expectedPath := "/internal/tasks/default/abc-def"
-			expectedName := "afc6f8271b8598ee04e359916e6c584a9bc3c520a11dd5244e3399346ac0d3a7"
+			expectedName := "prefix-afc6f8271b8598ee04e359916e6c584a9bc3c520a11dd5244e3399346ac0d3a7"
 			expectedBody := []byte(`{"type":"google.protobuf.Duration","body":"123.000s"}`)
 			tasks := taskqueue.GetTestable(ctx).GetScheduledTasks()
 			So(tasks, ShouldResemble, taskqueue.QueueData{
@@ -101,6 +115,7 @@
 			err = d.AddTask(ctx, &Task{
 				Payload:          &duration.Duration{Seconds: 123},
 				DeduplicationKey: "abc",
+				NamePrefix:       "prefix",
 			})
 			So(err, ShouldBeNil)
 
@@ -108,16 +123,35 @@
 			tasks = taskqueue.GetTestable(ctx).GetScheduledTasks()
 			So(len(tasks["default"]), ShouldResemble, 1)
 
-			// Execute the task.
-			req := httptest.NewRequest("POST", "http://example.com"+expectedPath, bytes.NewReader(expectedBody))
-			rw := httptest.NewRecorder()
-			r.ServeHTTP(rw, req)
-
-			// Executed.
-			So(calls, ShouldResemble, []proto.Message{
-				&duration.Duration{Seconds: 123},
+			Convey("Executed", func() {
+				// Execute the task.
+				So(runTasks(ctx), ShouldResemble, []int{200})
+				So(calls, ShouldResemble, []proto.Message{
+					&duration.Duration{Seconds: 123},
+				})
 			})
-			So(rw.Code, ShouldEqual, 200)
+
+			Convey("Deleted", func() {
+				So(d.DeleteTask(ctx, &Task{
+					Payload:          &duration.Duration{Seconds: 123},
+					DeduplicationKey: "abc",
+					NamePrefix:       "prefix",
+				}), ShouldBeNil)
+
+				// Did not execute any tasks.
+				So(runTasks(ctx), ShouldHaveLength, 0)
+				So(calls, ShouldHaveLength, 0)
+			})
+		})
+
+		Convey("Deleting unknown task returns nil", func() {
+			handler := func(c context.Context, payload proto.Message, execCount int) error { return nil }
+			d.RegisterTask(&duration.Duration{}, handler, "default", nil)
+
+			So(d.DeleteTask(ctx, &Task{
+				Payload:          &duration.Duration{Seconds: 123},
+				DeduplicationKey: "something",
+			}), ShouldBeNil)
 		})
 
 		Convey("Many tasks", func() {
@@ -166,6 +200,10 @@
 			}
 			So(len(delaysDefault), ShouldEqual, 100)
 			So(len(delaysAnotherQ), ShouldEqual, 100)
+
+			// Delete the tasks.
+			So(d.DeleteTask(ctx, t...), ShouldBeNil)
+			So(runTasks(ctx), ShouldHaveLength, 0)
 		})
 
 		Convey("Execution errors", func() {