Method ServeTask

worker/task.go:2224–2286

ServeTask is used to respond to a query.

(ctx context.Context, q *pb.Query)

// ServeTask is used to respond to a query.
func (w *grpcWorker) ServeTask(ctx context.Context, q *pb.Query) (*pb.Result, error) {
	// Manually extract trace context from gRPC metadata using the propagator
	// This ensures the trace context is properly extracted for cross-alpha tracing
	ctx = x.ExtractTraceContext(ctx)

	// Sanitize attr for span name to ensure valid UTF-8 for OTLP export
	safeAttr := x.SafeUTF8(q.Attr)
	ctx, span := otel.Tracer("").Start(ctx, "worker.ServeTask",
		trace.WithAttributes(attribute.String("predicate", safeAttr)))
	defer span.End()

	if ctx.Err() != nil {
		return nil, ctx.Err()
	}

	// It could be possible that the server isn't ready but a peer sends a
	// request. In that case we should check for the health here.
	if err := x.HealthCheck(); err != nil {
		return nil, err
	}

	gid, err := groups().BelongsToReadOnly(q.Attr, q.ReadTs)
	switch {
	case err != nil:
		return nil, err
	case gid == 0:
		return nil, errNonExistentTablet
	case gid != groups().groupId():
		return nil, errUnservedTablet
	}

	var numUids int
	if q.UidList != nil {
		numUids = len(q.UidList.Uids)
	}
	span.AddEvent("ServeTask details", trace.WithAttributes(
		attribute.String("attr", q.Attr),
		attribute.Int("num_uids", numUids),
		attribute.Int64("group_id", int64(gid))))

	if !groups().ServesGroup(gid) {
		return nil, errors.Errorf(
			"Temporary error, attr: %q groupId: %v Request sent to wrong server",
			x.ParseAttr(q.Attr), gid)
	}

	type reply struct {
		result *pb.Result
		err    error
	}
	c := make(chan reply, 1)
	go func() {
		result, err := processTask(ctx, q, gid)
		c <- reply{result, err}
	}()

	select {
	case <-ctx.Done():