Function TaskStatusOverNetwork

worker/queue.go:30–66

TaskStatusOverNetwork fetches the status of a task over the network. Alphas only know about the tasks created by them, but this function would fetch the task from the correct Alpha.

(ctx context.Context, req *pb.TaskStatusRequest,
)

// TaskStatusOverNetwork fetches the status of a task over the network. Alphas only know about the
// tasks created by them, but this function would fetch the task from the correct Alpha.
func TaskStatusOverNetwork(ctx context.Context, req *pb.TaskStatusRequest,
) (*pb.TaskStatusResponse, error) {
	// Extract Raft ID from Task ID.
	taskId := req.GetTaskId()
	if taskId == 0 {
		return nil, fmt.Errorf("invalid task ID: %#x", taskId)
	}
	raftId := taskId >> 32

	// Skip the network call if the required Alpha is me.
	myRaftId := State.WALstore.Uint(raftwal.RaftId)
	if raftId == myRaftId {
		worker := (*grpcWorker)(nil)
		return worker.TaskStatus(ctx, req)
	}

	// Find the Alpha with the required Raft ID.
	var addr string
	for _, group := range groups().state.GetGroups() {
		for _, member := range group.GetMembers() {
			if member.GetId() == raftId {
				addr = member.GetAddr()
			}
		}
	}
	if addr == "" {
		return nil, fmt.Errorf("the Alpha that served that task is not available")
	}

	// Send the request to the Alpha.
	pool, err := conn.GetPools().Get(addr)
	if err != nil {
		return nil, errors.Wrapf(err, "unable to reach the Alpha that served that task")
	}
	client := pb.NewWorkerClient(pool.Get())
	return client.TaskStatus(ctx, req)
}

// TaskStatus retrieves metadata for a given task ID.
func (*grpcWorker) TaskStatus(ctx context.Context, req *pb.TaskStatusRequest,