Google Git Google Git
Sign in
fuchsia/third_party/github.com/golang/appengine/refs/heads/upstream/taskqueue-migration-preview^!/.
commit
85a6c90f9acede2d26b175f153fa5a134967d68d
[log]
author
Chirag Gajjar <gajjarc@google.com>
Fri Aug 07 15:00:42 2026 +0530
committer
GitHub <noreply@github.com>
Fri Aug 07 15:00:42 2026 +0530
tree
a83443bfd47560678821ec4eb6638c2fd97c89ad
parent
af6b28078e3fe97b3e3964556a03a1ad73898416 [diff] 
Taskqueue migration preview (#410)

* v2: add Cloud Tasks transparent routing to v2/taskqueue for dual v1 and v2 module support

* taskqueue: implement QueueStats via Cloud Tasks v2beta3 GetQueue with stats read mask
diff --git a/taskqueue/cloudtask.go b/taskqueue/cloudtask.go
index 10e67d7..f7a8a54 100644
--- a/taskqueue/cloudtask.go
+++ b/taskqueue/cloudtask.go

@@ -17,8 +17,10 @@
 	pb "google.golang.org/appengine/internal/taskqueue"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/fieldmaskpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+
 	cloudtasks "cloud.google.com/go/cloudtasks/apiv2beta3"
 	taskspb "cloud.google.com/go/cloudtasks/apiv2beta3/cloudtaskspb"
 )
@@ -584,3 +586,42 @@
 	}
 	return fmt.Errorf("cloud tasks operation failed (%d): %s", code, msg)
 }
+
+func queueStatsInCloudTasks(ctx context.Context, queueNames []string) ([]QueueStatistics, error) {
+	client, err := cloudtasks.NewClient(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create cloudtasks client: %v", err)
+	}
+	defer client.Close()
+
+	qs := make([]QueueStatistics, len(queueNames))
+	for i, q := range queueNames {
+		fullQueueName, err := getQueuePath(ctx, q)
+		if err != nil {
+			return nil, err
+		}
+		req := &taskspb.GetQueueRequest{
+			Name: fullQueueName,
+			ReadMask: &fieldmaskpb.FieldMask{
+				Paths: []string{"stats"},
+			},
+		}
+		queue, err := client.GetQueue(ctx, req)
+		if err != nil {
+			return nil, err
+		}
+		if queue != nil && queue.Stats != nil {
+			qs[i] = QueueStatistics{
+				Tasks:           int(queue.Stats.TasksCount),
+				Executed1Minute: int(queue.Stats.ExecutedLastMinuteCount),
+				InFlight:        int(queue.Stats.ConcurrentDispatchesCount),
+				EnforcedRate:    queue.Stats.EffectiveExecutionRate,
+			}
+			if queue.Stats.OldestEstimatedArrivalTime != nil {
+				qs[i].OldestETA = queue.Stats.OldestEstimatedArrivalTime.AsTime()
+			}
+		}
+	}
+	return qs, nil
+}
+

diff --git a/taskqueue/taskqueue.go b/taskqueue/taskqueue.go
index ff09e1e..4843ac6 100644
--- a/taskqueue/taskqueue.go
+++ b/taskqueue/taskqueue.go

@@ -503,6 +503,9 @@
 
 // QueueStats retrieves statistics about queues.
 func QueueStats(c context.Context, queueNames []string) ([]QueueStatistics, error) {
+	if useCloudTasks() {
+		return queueStatsInCloudTasks(c, queueNames)
+	}
 	req := &pb.TaskQueueFetchQueueStatsRequest{
 		QueueName: make([][]byte, len(queueNames)),
 	}

diff --git a/v2/internal/transaction.go b/v2/internal/transaction.go
index 7198bac..56a9ff3 100644
--- a/v2/internal/transaction.go
+++ b/v2/internal/transaction.go

@@ -17,6 +17,23 @@
 	pb "google.golang.org/appengine/v2/internal/datastore"
 )
 
+// Transaction hooks for taskqueue outbox pattern
+var (
+	PostCommitHook func(ctx context.Context, handle uint64)
+	RollbackHook   func(handle uint64)
+)
+
+func TransactionFromContext(ctx context.Context) *pb.Transaction {
+	if t := transactionFromContext(ctx); t != nil {
+		return &t.transaction
+	}
+	return nil
+}
+
+func TransactionlessContext(ctx context.Context) context.Context {
+	return withTransaction(ctx, nil)
+}
+
 var transactionSetters = make(map[reflect.Type]reflect.Value)
 
 // RegisterTransactionSetter registers a function that sets transaction information
@@ -89,6 +106,9 @@
 		// Ignore the error return value, since we are already returning a non-nil
 		// error (or we're panicking).
 		Call(c, "datastore_v3", "Rollback", &t.transaction, &basepb.VoidProto{})
+		if RollbackHook != nil {
+			RollbackHook(t.transaction.GetHandle())
+		}
 	}()
 	if err := f(withTransaction(c, t)); err != nil {
 		return &t.transaction, err
@@ -111,5 +131,12 @@
 			return &t.transaction, ErrConcurrentTransaction
 		}
 	}
+	if err == nil {
+		if PostCommitHook != nil {
+			PostCommitHook(c, t.transaction.GetHandle())
+		}
+	} else if RollbackHook != nil {
+		RollbackHook(t.transaction.GetHandle())
+	}
 	return &t.transaction, err
 }

diff --git a/v2/taskqueue/cloudtask.go b/v2/taskqueue/cloudtask.go
new file mode 100644
index 0000000..8661894
--- /dev/null
+++ b/v2/taskqueue/cloudtask.go

@@ -0,0 +1,627 @@
+package taskqueue
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"google.golang.org/appengine/v2"
+	"google.golang.org/appengine/v2/datastore"
+	"google.golang.org/appengine/v2/internal"
+	pb "google.golang.org/appengine/v2/internal/taskqueue"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/fieldmaskpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+
+	cloudtasks "cloud.google.com/go/cloudtasks/apiv2beta3"
+	taskspb "cloud.google.com/go/cloudtasks/apiv2beta3/cloudtaskspb"
+)
+
+const (
+	maxTaskPayloadBytes   = 100 * 1024 // 100 KB max payload size for Cloud Tasks
+	maxTransactionalTasks = 5          // Maximum tasks allowed in a single Datastore transaction
+	batchCreateChunkSize  = 100        // Maximum tasks per BatchCreateTasks request
+	batchDeleteChunkSize  = 1000       // Maximum tasks per BatchDeleteTasks request
+
+	grpcNotFound      = 5
+	grpcAlreadyExists = 6
+	httpNotFound      = 404
+	httpAlreadyExists = 409
+)
+
+var (
+	taskNameRegex                = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
+	ErrTooManyTasksInTransaction = &internal.APIError{
+		Service: "taskqueue",
+		Detail:  "too many tasks in transaction",
+		Code:    int32(pb.TaskQueueServiceError_TOO_MANY_TASKS),
+	}
+)
+
+func useCloudTasks() bool {
+	v, _ := strconv.ParseBool(os.Getenv("APPENGINE_USE_CLOUDTASK_PUSH_QUEUE"))
+	return v
+}
+
+func newUnknownTaskError(detail string) error {
+	return &internal.APIError{
+		Service: "taskqueue",
+		Detail:  detail,
+		Code:    int32(pb.TaskQueueServiceError_UNKNOWN_TASK),
+	}
+}
+
+func isAlreadyExistsError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	return strings.Contains(msg, "AlreadyExists") || strings.Contains(msg, "already exists") || strings.Contains(msg, "409") || strings.Contains(msg, "Policy checks are unavailable")
+}
+
+func isUnimplementedError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	return strings.Contains(msg, "Unimplemented") || strings.Contains(msg, "unknown method") || strings.Contains(msg, "404")
+}
+
+func getQueuePath(ctx context.Context, queueName string) (string, error) {
+	if queueName == "" {
+		queueName = "default"
+	}
+	project := appengine.AppID(ctx)
+	if idx := strings.Index(project, "~"); idx != -1 {
+		project = project[idx+1:]
+	}
+	region, err := getRegion(ctx)
+	if err != nil {
+		return "", fmt.Errorf("failed to get region: %v", err)
+	}
+	return fmt.Sprintf("projects/%s/locations/%s/queues/%s", project, region, queueName), nil
+}
+
+func getRegion(ctx context.Context) (string, error) {
+	req, err := http.NewRequest("GET", "http://metadata.google.internal/computeMetadata/v1/instance/region", nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Metadata-Flavor", "Google")
+	resp, err := http.DefaultClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("metadata server returned status %d", resp.StatusCode)
+	}
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	parts := strings.Split(strings.TrimSpace(string(body)), "/")
+	if len(parts) == 0 {
+		return "", fmt.Errorf("invalid region format: %s", string(body))
+	}
+	return parts[len(parts)-1], nil
+}
+
+func sendTask(ctx context.Context, queueName string, taskName string, taskObj *taskspb.Task) (string, error) {
+	parent, err := getQueuePath(ctx, queueName)
+	if err != nil {
+		return "", err
+	}
+
+	client, err := cloudtasks.NewClient(ctx)
+	if err != nil {
+		return "", fmt.Errorf("failed to create cloudtasks client: %v", err)
+	}
+	defer client.Close()
+
+	req := &taskspb.CreateTaskRequest{
+		Parent: parent,
+		Task:   taskObj,
+	}
+
+	createdTask, err := client.CreateTask(ctx, req)
+	if err != nil {
+		if isAlreadyExistsError(err) {
+			return "", ErrTaskAlreadyAdded
+		}
+		return "", err
+	}
+	shortName := taskName
+	if createdTask != nil && createdTask.Name != "" {
+		if idx := strings.LastIndex(createdTask.Name, "/"); idx != -1 {
+			shortName = createdTask.Name[idx+1:]
+		} else {
+			shortName = createdTask.Name
+		}
+	}
+	return shortName, nil
+}
+
+func extractServiceFromHost(ctx context.Context, host string) string {
+	if host == "" {
+		if s := os.Getenv("GAE_SERVICE"); s != "" {
+			return s
+		}
+		return "default"
+	}
+
+	if idx := strings.Index(host, ":"); idx != -1 {
+		host = host[:idx]
+	}
+
+	project := appengine.AppID(ctx)
+	if idx := strings.Index(project, "~"); idx != -1 {
+		project = project[idx+1:]
+	}
+
+	pIdx := strings.Index(host, project)
+	if pIdx == -1 {
+		defaultHost := appengine.DefaultVersionHostname(ctx)
+		if host == defaultHost {
+			return "default"
+		}
+		return host
+	}
+
+	domainSuffix := host[pIdx:]
+	if host == domainSuffix {
+		return "default"
+	}
+
+	suffixes := []string{
+		"." + domainSuffix,
+		"-dot-" + domainSuffix,
+	}
+	stripped := host
+	for _, suffix := range suffixes {
+		if strings.HasSuffix(stripped, suffix) {
+			stripped = stripped[:len(stripped)-len(suffix)]
+			break
+		}
+	}
+
+	if stripped == host {
+		return host
+	}
+
+	stripped = strings.ReplaceAll(stripped, "-dot-", ".")
+	parts := strings.Split(stripped, ".")
+	if len(parts) > 0 {
+		return parts[len(parts)-1]
+	}
+	return "default"
+}
+
+func buildCloudTaskProto(ctx context.Context, queueName string, task *Task) (*taskspb.Task, string, error) {
+	if task.Name != "" {
+		if !taskNameRegex.MatchString(task.Name) {
+			return nil, "", fmt.Errorf("taskqueue: invalid task name %q", task.Name)
+		}
+	}
+
+	if len(task.Payload) > maxTaskPayloadBytes {
+		return nil, "", fmt.Errorf("taskqueue: task too large (%d bytes)", len(task.Payload))
+	}
+
+	queuePath, err := getQueuePath(ctx, queueName)
+	if err != nil {
+		return nil, "", err
+	}
+
+	taskName := task.Name
+	var fullTaskName string
+	if taskName != "" {
+		fullTaskName = fmt.Sprintf("%s/tasks/%s", queuePath, taskName)
+	}
+
+	path := task.Path
+	if path == "" {
+		path = "/_ah/queue/" + queueName
+	}
+
+	headers := make(map[string]string)
+	for k, vs := range task.Header {
+		if len(vs) > 0 {
+			headers[k] = vs[0]
+		}
+	}
+
+	if _, ok := headers["Content-Type"]; !ok {
+		headers["Content-Type"] = "application/octet-stream"
+	}
+	if _, ok := headers["X-AppEngine-QueueName"]; !ok {
+		headers["X-AppEngine-QueueName"] = queueName
+	}
+	if taskName != "" {
+		if _, ok := headers["X-AppEngine-TaskName"]; !ok {
+			headers["X-AppEngine-TaskName"] = taskName
+		}
+	}
+
+	targetService := extractServiceFromHost(ctx, headers["Host"])
+	var routing *taskspb.AppEngineRouting
+	if targetService != "" {
+		routing = &taskspb.AppEngineRouting{
+			Service: targetService,
+		}
+	}
+	delete(headers, "Host")
+	ae := &taskspb.AppEngineHttpRequest{
+		RelativeUri:      path,
+		Headers:          headers,
+		Body:             task.Payload,
+		AppEngineRouting: routing,
+	}
+	if code, ok := taskspb.HttpMethod_value[task.method()]; ok {
+		ae.HttpMethod = taskspb.HttpMethod(code)
+	}
+
+	taskObj := &taskspb.Task{
+		Name: fullTaskName,
+		PayloadType: &taskspb.Task_AppEngineHttpRequest{
+			AppEngineHttpRequest: ae,
+		},
+	}
+
+	if !task.ETA.IsZero() {
+		taskObj.ScheduleTime = timestamppb.New(task.ETA)
+	} else if task.Delay > 0 {
+		taskObj.ScheduleTime = timestamppb.New(time.Now().Add(task.Delay))
+	}
+
+	if task.RetryOptions != nil {
+		rc := &taskspb.RetryConfig{}
+		hasRC := false
+		if task.RetryOptions.RetryLimit > 0 {
+			rc.MaxAttempts = task.RetryOptions.RetryLimit
+			hasRC = true
+		}
+		if task.RetryOptions.AgeLimit > 0 {
+			rc.MaxRetryDuration = durationpb.New(task.RetryOptions.AgeLimit)
+			hasRC = true
+		}
+		if task.RetryOptions.MinBackoff > 0 {
+			rc.MinBackoff = durationpb.New(task.RetryOptions.MinBackoff)
+			hasRC = true
+		}
+		if task.RetryOptions.MaxBackoff > 0 {
+			rc.MaxBackoff = durationpb.New(task.RetryOptions.MaxBackoff)
+			hasRC = true
+		}
+		if task.RetryOptions.MaxDoublings > 0 || (task.RetryOptions.MaxDoublings == 0 && task.RetryOptions.ApplyZeroMaxDoublings) {
+			rc.MaxDoublings = task.RetryOptions.MaxDoublings
+			hasRC = true
+		}
+		if hasRC {
+			taskObj.RetryConfig = rc
+		}
+	}
+
+	return taskObj, taskName, nil
+}
+
+func addInCloudTasks(ctx context.Context, task *Task, queueName string) (*Task, error) {
+	if queueName == "" {
+		queueName = "default"
+	}
+
+	taskObj, taskName, err := buildCloudTaskProto(ctx, queueName, task)
+	if err != nil {
+		return nil, err
+	}
+
+	// In App Engine Datastore, external HTTP/gRPC Cloud Tasks RPCs cannot participate
+	// in Datastore 2PC transactions. If we are running inside an active Datastore transaction,
+	// we stage the task as a _AE_PendingCloudTask entity in Datastore under the transaction.
+	// When the transaction commits, PostCommitHook dispatches the staged task to Cloud Tasks.
+	if t := internal.TransactionFromContext(ctx); t != nil {
+		handle := t.GetHandle()
+		pendingTasksMu.Lock()
+		if len(pendingTasks[handle]) >= maxTransactionalTasks {
+			pendingTasksMu.Unlock()
+			return nil, ErrTooManyTasksInTransaction
+		}
+		pendingTasksMu.Unlock()
+
+		protoBytes, err := proto.Marshal(taskObj)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal proto for transactional task: %v", err)
+		}
+		key := datastore.NewIncompleteKey(ctx, "_AE_PendingCloudTask", nil)
+		pendingTask := &PendingCloudTask{
+			QueueName:        queueName,
+			CloudTaskName:    taskName,
+			CloudTaskPayload: string(protoBytes),
+			Created:          time.Now(),
+			Status:           "PENDING",
+			RetryCount:       0,
+			LastError:        "",
+			HandledBySweeper: false,
+			SdkLang:          "GO",
+		}
+		key, err = datastore.Put(ctx, key, pendingTask)
+		if err != nil {
+			return nil, fmt.Errorf("failed to save transactional task to Datastore: %v", err)
+		}
+
+		pendingTasksMu.Lock()
+		pendingTasks[handle] = append(pendingTasks[handle], key.Encode())
+		pendingTasksMu.Unlock()
+
+		resultTask := *task
+		resultTask.Name = taskName
+		resultTask.Method = task.method()
+		return &resultTask, nil
+	}
+
+	assignedName, err := sendTask(ctx, queueName, taskName, taskObj)
+	if err != nil {
+		return nil, err
+	}
+
+	resultTask := *task
+	resultTask.Name = assignedName
+	resultTask.Method = task.method()
+	return &resultTask, nil
+}
+
+func addMultiInCloudTasks(ctx context.Context, tasks []*Task, queueName string) ([]*Task, error) {
+	// If AddMulti is called inside a Datastore transaction, each task in the batch
+	// is transactionally staged in Datastore via addInCloudTasks so that all tasks
+	// commit atomically with the Datastore transaction.
+	if t := internal.TransactionFromContext(ctx); t != nil {
+		handle := t.GetHandle()
+		pendingTasksMu.Lock()
+		if len(pendingTasks[handle])+len(tasks) > maxTransactionalTasks {
+			pendingTasksMu.Unlock()
+			return nil, ErrTooManyTasksInTransaction
+		}
+		pendingTasksMu.Unlock()
+
+		me, any := make(appengine.MultiError, len(tasks)), false
+		results := make([]*Task, len(tasks))
+		for i, task := range tasks {
+			res, err := addInCloudTasks(ctx, task, queueName)
+			if err != nil {
+				me[i] = err
+				any = true
+			} else {
+				results[i] = res
+			}
+		}
+		if any {
+			return results, me
+		}
+		return results, nil
+	}
+
+	fullQueueName, err := getQueuePath(ctx, queueName)
+	if err != nil {
+		return nil, err
+	}
+
+	me, any := make(appengine.MultiError, len(tasks)), false
+	results := make([]*Task, len(tasks))
+
+	client, err := cloudtasks.NewClient(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create cloudtasks client: %v", err)
+	}
+	defer client.Close()
+
+	chunkSize := batchCreateChunkSize
+	for chunkStart := 0; chunkStart < len(tasks); chunkStart += chunkSize {
+		chunkEnd := chunkStart + chunkSize
+		if chunkEnd > len(tasks) {
+			chunkEnd = len(tasks)
+		}
+		chunkTasks := tasks[chunkStart:chunkEnd]
+
+		createReqs := make([]*taskspb.CreateTaskRequest, 0, len(chunkTasks))
+		for i, t := range chunkTasks {
+			taskObj, taskName, err := buildCloudTaskProto(ctx, queueName, t)
+			if err != nil {
+				me[chunkStart+i] = err
+				any = true
+				continue
+			}
+			results[chunkStart+i] = new(Task)
+			*results[chunkStart+i] = *t
+			results[chunkStart+i].Name = taskName
+			results[chunkStart+i].Method = t.method()
+
+			createReqs = append(createReqs, &taskspb.CreateTaskRequest{
+				Parent: fullQueueName,
+				Task:   taskObj,
+			})
+		}
+		if len(createReqs) == 0 {
+			continue
+		}
+
+		batchReq := &taskspb.BatchCreateTasksRequest{
+			Parent:   fullQueueName,
+			Requests: createReqs,
+		}
+
+		op, err := client.BatchCreateTasks(ctx, batchReq)
+		if err != nil {
+			if isUnimplementedError(err) {
+				for i, t := range chunkTasks {
+					if me[chunkStart+i] != nil {
+						continue
+					}
+					res, err := addInCloudTasks(ctx, t, queueName)
+					if err != nil {
+						me[chunkStart+i] = err
+						any = true
+					} else {
+						results[chunkStart+i] = res
+					}
+				}
+			} else {
+				for i := range chunkTasks {
+					if me[chunkStart+i] == nil {
+						me[chunkStart+i] = err
+						any = true
+					}
+				}
+			}
+		} else if op != nil {
+			meta, _ := op.Metadata()
+			resp, _ := op.Wait(ctx)
+			for i := range chunkTasks {
+				if meta != nil && meta.FailedRequests != nil {
+					if st, failed := meta.FailedRequests[int32(i)]; failed && st != nil && st.Code != 0 {
+						me[chunkStart+i] = mapOperationErrorCode(int(st.Code), st.Message, false)
+						any = true
+						continue
+					}
+				}
+				if resp != nil && i < len(resp.Tasks) && resp.Tasks[i] != nil {
+					createdTask := resp.Tasks[i]
+					if createdTask.Name != "" && results[chunkStart+i] != nil {
+						if idx := strings.LastIndex(createdTask.Name, "/"); idx != -1 {
+							results[chunkStart+i].Name = createdTask.Name[idx+1:]
+						} else {
+							results[chunkStart+i].Name = createdTask.Name
+						}
+					}
+				}
+			}
+		}
+	}
+
+	if any {
+		return results, me
+	}
+	return results, nil
+}
+
+func deleteMultiInCloudTasks(ctx context.Context, tasks []*Task, queueName string) error {
+	fullQueueName, err := getQueuePath(ctx, queueName)
+	if err != nil {
+		return err
+	}
+
+	client, err := cloudtasks.NewClient(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create cloudtasks client: %v", err)
+	}
+	defer client.Close()
+
+	me, any := make(appengine.MultiError, len(tasks)), false
+
+	chunkSize := batchDeleteChunkSize
+	for chunkStart := 0; chunkStart < len(tasks); chunkStart += chunkSize {
+		chunkEnd := chunkStart + chunkSize
+		if chunkEnd > len(tasks) {
+			chunkEnd = len(tasks)
+		}
+		chunkTasks := tasks[chunkStart:chunkEnd]
+
+		names := make([]string, len(chunkTasks))
+		for i, t := range chunkTasks {
+			names[i] = fmt.Sprintf("%s/tasks/%s", fullQueueName, t.Name)
+		}
+
+		batchReq := &taskspb.BatchDeleteTasksRequest{
+			Parent: fullQueueName,
+			Names:  names,
+		}
+
+		op, err := client.BatchDeleteTasks(ctx, batchReq)
+		if err != nil {
+			for i := range chunkTasks {
+				me[chunkStart+i] = err
+				any = true
+			}
+		} else if op != nil {
+			meta, _ := op.Metadata()
+			for i := range chunkTasks {
+				if meta != nil && meta.FailedRequests != nil {
+					if st, failed := meta.FailedRequests[int32(i)]; failed && st != nil && st.Code != 0 {
+						me[chunkStart+i] = mapOperationErrorCode(int(st.Code), st.Message, true)
+						any = true
+					}
+				}
+			}
+		}
+	}
+
+	if any {
+		return me
+	}
+	return nil
+}
+
+
+
+func mapOperationErrorCode(code int, msg string, isDelete bool) error {
+	lowerMsg := strings.ToLower(msg)
+	isNotFound := code == grpcNotFound || code == httpNotFound || strings.Contains(lowerMsg, "not found") || strings.Contains(lowerMsg, "unknown")
+	isAlreadyExists := code == grpcAlreadyExists || code == httpAlreadyExists || strings.Contains(lowerMsg, "already exists")
+
+	if isDelete && isNotFound {
+		return newUnknownTaskError(msg)
+	}
+	if isAlreadyExists || (isNotFound && strings.Contains(lowerMsg, "requested entity was not found")) {
+		return ErrTaskAlreadyAdded
+	}
+	if isNotFound {
+		return newUnknownTaskError(msg)
+	}
+	return fmt.Errorf("cloud tasks operation failed (%d): %s", code, msg)
+}
+
+func queueStatsInCloudTasks(ctx context.Context, queueNames []string) ([]QueueStatistics, error) {
+	client, err := cloudtasks.NewClient(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create cloudtasks client: %v", err)
+	}
+	defer client.Close()
+
+	qs := make([]QueueStatistics, len(queueNames))
+	for i, q := range queueNames {
+		fullQueueName, err := getQueuePath(ctx, q)
+		if err != nil {
+			return nil, err
+		}
+		req := &taskspb.GetQueueRequest{
+			Name: fullQueueName,
+			ReadMask: &fieldmaskpb.FieldMask{
+				Paths: []string{"stats"},
+			},
+		}
+		queue, err := client.GetQueue(ctx, req)
+		if err != nil {
+			return nil, err
+		}
+		if queue != nil && queue.Stats != nil {
+			qs[i] = QueueStatistics{
+				Tasks:           int(queue.Stats.TasksCount),
+				Executed1Minute: int(queue.Stats.ExecutedLastMinuteCount),
+				InFlight:        int(queue.Stats.ConcurrentDispatchesCount),
+				EnforcedRate:    queue.Stats.EffectiveExecutionRate,
+			}
+			if queue.Stats.OldestEstimatedArrivalTime != nil {
+				qs[i].OldestETA = queue.Stats.OldestEstimatedArrivalTime.AsTime()
+			}
+		}
+	}
+	return qs, nil
+}
+

diff --git a/v2/taskqueue/cloudtask_transactional.go b/v2/taskqueue/cloudtask_transactional.go
new file mode 100644
index 0000000..9745f77
--- /dev/null
+++ b/v2/taskqueue/cloudtask_transactional.go

@@ -0,0 +1,242 @@
+package taskqueue
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"google.golang.org/appengine/v2"
+	"google.golang.org/appengine/v2/datastore"
+	"google.golang.org/appengine/v2/internal"
+	"google.golang.org/protobuf/proto"
+
+	taskspb "cloud.google.com/go/cloudtasks/apiv2beta3/cloudtaskspb"
+)
+
+const (
+	statusPending       = "PENDING"
+	statusProcessing    = "PROCESSING"
+	statusFailed        = "FAILED"
+	statusDone          = "DONE"
+	statusAlreadyExists = "ALREADY_EXISTS"
+
+	lockDuration        = 60 * time.Second
+	fastPathGracePeriod = 60 * time.Second
+	maxSweeperRetries   = 5
+	maxLastErrorLength  = 500
+)
+
+type PendingCloudTask struct {
+	QueueName        string    `datastore:"queue_name"`
+	CloudTaskName    string    `datastore:"cloud_task_name"`
+	CloudTaskPayload string    `datastore:"cloud_task_payload,noindex"`
+	Created          time.Time `datastore:"created"`
+	Status           string    `datastore:"status"`
+	LockExpires      time.Time `datastore:"lock_expires"`
+	RetryCount       int64     `datastore:"retry_count"`
+	LastError        string    `datastore:"last_error,noindex"`
+	HandledBySweeper bool      `datastore:"handled_by_sweeper"`
+	SdkLang          string    `datastore:"sdk_lang"`
+}
+
+var (
+	pendingTasksMu sync.Mutex
+	pendingTasks   = make(map[uint64][]string) // transaction handle -> list of urlsafe keys
+)
+
+func init() {
+	internal.PostCommitHook = func(ctx context.Context, handle uint64) {
+		go dispatchPendingTasks(ctx, handle)
+	}
+	internal.RollbackHook = func(handle uint64) {
+		cleanupPendingTasks(handle)
+	}
+	http.HandleFunc("/_ah/cloudtask/sweep", handleSweep)
+}
+
+func cleanupPendingTasks(handle uint64) {
+	pendingTasksMu.Lock()
+	delete(pendingTasks, handle)
+	pendingTasksMu.Unlock()
+}
+
+type noCancelContext struct {
+	context.Context
+}
+
+func (c *noCancelContext) Deadline() (deadline time.Time, ok bool) {
+	return time.Time{}, false
+}
+
+func (c *noCancelContext) Done() <-chan struct{} {
+	return nil
+}
+
+func (c *noCancelContext) Err() error {
+	return nil
+}
+
+func logErrorf(ctx context.Context, format string, v ...interface{}) {
+	log.Printf("ERROR: "+format, v...)
+}
+
+func dispatchPendingTasks(ctx context.Context, handle uint64) {
+	pendingTasksMu.Lock()
+	urlsafeKeys, ok := pendingTasks[handle]
+	if ok {
+		delete(pendingTasks, handle)
+	}
+	pendingTasksMu.Unlock()
+
+	if !ok || len(urlsafeKeys) == 0 {
+		return
+	}
+
+	noCancelCtx := &noCancelContext{Context: internal.TransactionlessContext(ctx)}
+
+	for _, urlsafeKey := range urlsafeKeys {
+		key, err := datastore.DecodeKey(urlsafeKey)
+		if err != nil {
+			logErrorf(ctx, "Failed to decode pending task key: %v", err)
+			continue
+		}
+
+		var taskEntity PendingCloudTask
+		err = datastore.Get(noCancelCtx, key, &taskEntity)
+		if err != nil {
+			logErrorf(ctx, "Failed to get pending task from Datastore: %v", err)
+			continue
+		}
+
+		now := time.Now()
+		taskEntity.Status = statusProcessing
+		taskEntity.LockExpires = now.Add(lockDuration)
+		taskEntity.HandledBySweeper = false
+		if _, err := datastore.Put(noCancelCtx, key, &taskEntity); err != nil {
+			logErrorf(ctx, "Failed to acquire lock in fast-path for task %s: %v", taskEntity.CloudTaskName, err)
+			continue
+		}
+
+		var taskObj taskspb.Task
+		if err := proto.Unmarshal([]byte(taskEntity.CloudTaskPayload), &taskObj); err != nil {
+			logErrorf(ctx, "Failed to unmarshal pending task proto: %v", err)
+			continue
+		}
+		_, err = sendTask(noCancelCtx, taskEntity.QueueName, taskEntity.CloudTaskName, &taskObj)
+		if err != nil {
+			if err == ErrTaskAlreadyAdded {
+				datastore.Delete(noCancelCtx, key)
+				continue
+			}
+			logErrorf(ctx, "Failed to dispatch task %s to queue %s: %v", taskEntity.CloudTaskName, taskEntity.QueueName, err)
+			taskEntity.RetryCount++
+			taskEntity.LastError = err.Error()
+			if len(taskEntity.LastError) > maxLastErrorLength {
+				taskEntity.LastError = taskEntity.LastError[:maxLastErrorLength]
+			}
+			taskEntity.Status = statusPending
+			datastore.Put(noCancelCtx, key, &taskEntity)
+			continue
+		}
+
+		err = datastore.Delete(noCancelCtx, key)
+		if err != nil {
+			logErrorf(ctx, "Failed to delete pending task %s from Datastore: %v", taskEntity.CloudTaskName, err)
+		}
+	}
+}
+
+func sweep(ctx context.Context) error {
+	query := datastore.NewQuery("_AE_PendingCloudTask")
+	var tasks []PendingCloudTask
+	keys, err := query.GetAll(ctx, &tasks)
+	if err != nil {
+		return fmt.Errorf("failed to query _AE_PendingCloudTask: %v", err)
+	}
+
+	now := time.Now()
+	count := 0
+	for i, key := range keys {
+		task := tasks[i]
+		if task.Status == statusDone || task.Status == statusAlreadyExists {
+			continue
+		}
+		if task.Status == statusProcessing {
+			if !task.LockExpires.IsZero() && now.Before(task.LockExpires) {
+				continue // Still actively processing and lock valid
+			} else if task.LockExpires.IsZero() {
+				continue // Assume lock valid if just started
+			}
+		} else if task.Status == statusPending || task.Status == "" {
+			if !task.Created.IsZero() && now.Sub(task.Created) < fastPathGracePeriod {
+				continue // Give fast-path grace period to dispatch post-commit
+			}
+		} else if task.Status == statusFailed && task.RetryCount >= maxSweeperRetries {
+			continue // Exceeded max sweeper retries
+		}
+
+		// Acquire lock
+		task.Status = statusProcessing
+		task.LockExpires = now.Add(lockDuration)
+		task.HandledBySweeper = true
+		if _, err := datastore.Put(ctx, key, &task); err != nil {
+			logErrorf(ctx, "Sweeper failed to acquire lock for task %s: %v", task.CloudTaskName, err)
+			continue
+		}
+
+		var taskObj taskspb.Task
+		if err := proto.Unmarshal([]byte(task.CloudTaskPayload), &taskObj); err != nil {
+			logErrorf(ctx, "Sweeper failed to unmarshal pending task proto: %v", err)
+			continue
+		}
+		_, err := sendTask(ctx, task.QueueName, task.CloudTaskName, &taskObj)
+		if err != nil && err != ErrTaskAlreadyAdded {
+			logErrorf(ctx, "Sweeper failed to dispatch task %s: %v", task.CloudTaskName, err)
+			task.RetryCount++
+			task.LastError = err.Error()
+			if len(task.LastError) > maxLastErrorLength {
+				task.LastError = task.LastError[:maxLastErrorLength]
+			}
+			if task.RetryCount >= maxSweeperRetries {
+				task.Status = statusFailed
+				task.LockExpires = time.Time{}
+			} else {
+				task.Status = statusPending
+				task.LockExpires = time.Time{}
+			}
+			if _, putErr := datastore.Put(ctx, key, &task); putErr != nil {
+				logErrorf(ctx, "Sweeper failed to record error state for task %s: %v", task.CloudTaskName, putErr)
+			}
+			continue
+		}
+
+		if err := datastore.Delete(ctx, key); err != nil {
+			logErrorf(ctx, "Sweeper failed to delete entity %s: %v", task.CloudTaskName, err)
+		}
+		count++
+	}
+
+	log.Printf("Cloud Tasks sweeper processed %d tasks.", count)
+	return nil
+}
+
+func handleSweep(w http.ResponseWriter, r *http.Request) {
+	isCron := strings.EqualFold(r.Header.Get("X-AppEngine-Cron"), "true") || strings.EqualFold(r.Header.Get("X-Appengine-Cron"), "true")
+	if !isCron && !appengine.IsDevAppServer() {
+		http.Error(w, "Access denied: endpoint only accessible via App Engine Cron.", http.StatusForbidden)
+		return
+	}
+	ctx := appengine.NewContext(r)
+	if err := sweep(ctx); err != nil {
+		logErrorf(ctx, "Sweeper failed: %v", err)
+		http.Error(w, fmt.Sprintf("Sweeper failed: %v", err), http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "text/plain")
+	w.WriteHeader(http.StatusOK)
+	w.Write([]byte("Sweeper completed successfully.\n"))
+}

diff --git a/v2/taskqueue/taskqueue.go b/v2/taskqueue/taskqueue.go
index 8bbd85a..dcf2c3e 100644
--- a/v2/taskqueue/taskqueue.go
+++ b/v2/taskqueue/taskqueue.go

@@ -283,6 +283,9 @@
 // Add returns an equivalent Task with defaults filled in, including setting
 // the task's Name field to the chosen name if the original was empty.
 func Add(c context.Context, task *Task, queueName string) (*Task, error) {
+	if useCloudTasks() && task.Method != "PULL" {
+		return addInCloudTasks(c, task, queueName)
+	}
 	req, err := newAddReq(c, task, queueName)
 	if err != nil {
 		return nil, err
@@ -309,6 +312,11 @@
 // each task's Name field to the chosen name if the original was empty.
 // If a given task is badly formed or could not be added, an appengine.MultiError is returned.
 func AddMulti(c context.Context, tasks []*Task, queueName string) ([]*Task, error) {
+	if useCloudTasks() {
+		if len(tasks) > 0 && tasks[0].Method != "PULL" {
+			return addMultiInCloudTasks(c, tasks, queueName)
+		}
+	}
 	req := &pb.TaskQueueBulkAddRequest{
 		AddRequest: make([]*pb.TaskQueueAddRequest, len(tasks)),
 	}
@@ -367,6 +375,11 @@
 // Each task is deleted independently; one may fail to delete while the others
 // are successfully deleted.
 func DeleteMulti(c context.Context, tasks []*Task, queueName string) error {
+	if useCloudTasks() {
+		if len(tasks) > 0 && tasks[0].Method != "PULL" {
+			return deleteMultiInCloudTasks(c, tasks, queueName)
+		}
+	}
 	taskNames := make([][]byte, len(tasks))
 	for i, t := range tasks {
 		taskNames[i] = []byte(t.Name)
@@ -490,6 +503,9 @@
 
 // QueueStats retrieves statistics about queues.
 func QueueStats(c context.Context, queueNames []string) ([]QueueStatistics, error) {
+	if useCloudTasks() {
+		return queueStatsInCloudTasks(c, queueNames)
+	}
 	req := &pb.TaskQueueFetchQueueStatsRequest{
 		QueueName: make([][]byte, len(queueNames)),
 	}