// Copyright 2016 The Bazel Authors. All rights reserved.
//
// 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,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.devtools.build.lib.server;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.net.InetAddresses;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.common.util.concurrent.Uninterruptibles;
import com.google.devtools.build.lib.bugreport.BugReport;
import com.google.devtools.build.lib.clock.Clock;
import com.google.devtools.build.lib.runtime.BlazeCommandResult;
import com.google.devtools.build.lib.runtime.BlazeRuntime;
import com.google.devtools.build.lib.runtime.CommandDispatcher;
import com.google.devtools.build.lib.runtime.CommandDispatcher.LockingMode;
import com.google.devtools.build.lib.runtime.proto.InvocationPolicyOuterClass.InvocationPolicy;
import com.google.devtools.build.lib.server.CommandManager.RunningCommand;
import com.google.devtools.build.lib.server.CommandProtos.CancelRequest;
import com.google.devtools.build.lib.server.CommandProtos.CancelResponse;
import com.google.devtools.build.lib.server.CommandProtos.PingRequest;
import com.google.devtools.build.lib.server.CommandProtos.PingResponse;
import com.google.devtools.build.lib.server.CommandProtos.RunRequest;
import com.google.devtools.build.lib.server.CommandProtos.RunResponse;
import com.google.devtools.build.lib.server.CommandProtos.ServerInfo;
import com.google.devtools.build.lib.server.CommandProtos.StartupOption;
import com.google.devtools.build.lib.server.FailureDetails.Interrupted;
import com.google.devtools.build.lib.util.ExitCode;
import com.google.devtools.build.lib.util.FailureDetailUtil;
import com.google.devtools.build.lib.util.Pair;
import com.google.devtools.build.lib.util.io.OutErr;
import com.google.devtools.build.lib.vfs.FileSystemUtils;
import com.google.devtools.build.lib.vfs.Path;
import com.google.devtools.common.options.InvocationPolicyParser;
import com.google.devtools.common.options.OptionsParsingException;
import com.google.protobuf.ByteString;
import io.grpc.Context;
import io.grpc.Server;
import io.grpc.StatusRuntimeException;
import io.grpc.netty.NettyServerBuilder;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

/**
 * gRPC server class.
 *
 * <p>Only this class should depend on gRPC so that we only need to exclude this during
 * bootstrapping.
 *
 * <p>This class is a little complicated and rich in multithreading, so an explanation of its
 * innards follows.
 *
 * <p>We use the direct executor for gRPC so that it calls our methods directly on its event handler
 * threads (which it creates itself). This is acceptable for {@code ping()} and {@code cancel()}
 * because they run very quickly. For {@code run()}, we transfer the call to our own threads in
 * {@code commandExecutorPool}. We do this instead of setting an executor on the server object
 * because gRPC insists on serializing calls within a single RPC call, which means that the Runnable
 * passed to {@code setOnReadyHandler} doesn't get called while the main RPC method is running,
 * which means we can't use flow control, which we need so that gRPC doesn't buffer an unbounded
 * amount of outgoing data.
 *
 * <p>Two threads are spawned for each command: one that handles the command in {@code
 * commandExecutorPool} and one that streams the result back to the client in {@code
 * streamExecutorPool}.
 *
 * <p>In addition to these threads, we maintain one extra thread for handling the server timeout and
 * an interrupt watcher thread is started for each interrupt request that logs if it takes too long
 * to take effect.
 *
 * <p>Each running RPC has a UUID associated with it that is used to identify it when a client wants
 * to cancel it. Cancellation is done by the client sending the server a {@code cancel()} RPC call
 * which results in the main thread of the command being interrupted.
 */
public class GrpcServerImpl extends CommandServerGrpc.CommandServerImplBase implements RPCServer {
  private static final Logger logger = Logger.getLogger(GrpcServerImpl.class.getName());
  private final boolean shutdownOnLowSysMem;

  /**
   * Factory class. Instantiated by reflection.
   *
   * <p>Used so that method calls using reflection are as simple as possible.
   */
  public static class Factory implements RPCServer.Factory {
    @Override
    public RPCServer create(
        CommandDispatcher dispatcher,
        Clock clock,
        int port,
        Path serverDirectory,
        int maxIdleSeconds,
        boolean shutdownOnLowSysMem,
        boolean idleServerTasks)
        throws IOException {
      SecureRandom random = new SecureRandom();
      return new GrpcServerImpl(
          dispatcher,
          clock,
          port,
          generateCookie(random, 16),
          generateCookie(random, 16),
          serverDirectory,
          maxIdleSeconds,
          shutdownOnLowSysMem,
          idleServerTasks);
    }
  }

  @VisibleForTesting
  enum StreamType {
    STDOUT,
    STDERR,
  }

  /**
   * A wrapper for {@link StreamObserver} that blocks on {@link #onNext} calls if the underlying
   * observer is not ready.
   *
   * <p>It does not react to the interrupt flag in order to allow Bazel to complete the current
   * command while printing output as well as sending the final exit code to the client. However, it
   * maintains the interrupt flag if it is already set.
   */
  // TODO(ulfjack): Move FlowControl and its tests to top-level classes.
  @VisibleForTesting
  static class BlockingStreamObserver<T> {
    private final ServerCallStreamObserver<T> observer;

    BlockingStreamObserver(ServerCallStreamObserver<T> observer) {
      this.observer = observer;
      this.observer.setOnReadyHandler(this::notifyWaiters);
      this.observer.setOnCancelHandler(this::notifyWaiters);
    }

    private synchronized void notifyWaiters() {
      // This class does not restrict the number of concurrent calls to onNext, so we call notifyAll
      // here. In practice we'll usually only see one concurrent call; the ExperimentalEventHandler
      // uses synchronization to prevent multiple concurrent calls, but let's not rely on that here.
      notifyAll();
    }

    public synchronized void onNext(T response) {
      boolean interrupted = false;
      while (!observer.isReady() && !observer.isCancelled()) {
        try {
          wait();
        } catch (InterruptedException e) {
          // We intentionally do not break or return here. The interrupt signal can be due the user
          // pressing ctrl-c: it can take Bazel a while to shut down (e.g., it is not currently
          // possible to interrupt persistent workers), and we must allow it to continue printing
          // output until the current operation comes to a finish.
          interrupted = true;
        }
      }
      try {
        // According to the documentation, if onNext is called in a canceled stream, it will be
        // silently ignored.
        observer.onNext(response);
      } finally {
        // onNext does not specify whether it can throw unchecked exceptions. We use a finally block
        // here to make sure that the interrupt bit is not lost even if it does.
        if (interrupted || observer.isCancelled()) {
          Thread.currentThread().interrupt();
        }
      }
    }

    public void onCompleted() {
      observer.onCompleted();
    }
  }

  /**
   * An output stream that forwards the data written to it over the gRPC command stream.
   *
   * <p>Note that wraping this class with a {@code Channel} can cause a deadlock if there is an
   * {@link OutputStream} in between that synchronizes both on {@code #close()} and {@code #write()}
   * because then if an interrupt happens in {@code FlowControl#onNext}, the thread on which {@code
   * interrupt()} was called will wait until the {@code Channel} closes itself while holding a lock
   * for interrupting the thread on which {@code FlowControl#onNext} is being executed and that
   * thread will hold a lock that is needed for the {@code Channel} to be closed and call {@code
   * interrupt()} in {@code FlowControl#onNext}, which will in turn try to acquire the interrupt
   * lock.
   */
  private static class RpcOutputStream extends OutputStream {
    private static final int CHUNK_SIZE = 8192;

    // Store commandId and responseCookie as ByteStrings to avoid String -> UTF8 bytes conversion
    // for each serialized chunk of output.
    private final ByteString commandIdBytes;
    private final ByteString responseCookieBytes;

    private final StreamType type;
    private final BlockingStreamObserver<RunResponse> observer;

    RpcOutputStream(
        String commandId,
        String responseCookie,
        StreamType type,
        BlockingStreamObserver<RunResponse> observer) {
      this.commandIdBytes = ByteString.copyFromUtf8(commandId);
      this.responseCookieBytes = ByteString.copyFromUtf8(responseCookie);
      this.type = type;
      this.observer = observer;
    }

    @Override
    public void write(byte[] b, int off, int inlen) throws IOException {
      for (int i = 0; i < inlen; i += CHUNK_SIZE) {
        ByteString input = ByteString.copyFrom(b, off + i, Math.min(CHUNK_SIZE, inlen - i));
        RunResponse.Builder response = RunResponse
            .newBuilder()
            .setCookieBytes(responseCookieBytes)
            .setCommandIdBytes(commandIdBytes);

        switch (type) {
          case STDOUT: response.setStandardOutput(input); break;
          case STDERR: response.setStandardError(input); break;
          default: throw new IllegalStateException();
        }

        try {
          // This can block waiting for the client to read the available data.
          observer.onNext(response.build());
        } catch (StatusRuntimeException e) {
          // I am not sure whether there are any circumstances under which this call could throw an
          // exception, but I'd rather it be logged than that we crash silently. The documentation
          // only says that onNext does not throw a CancelledException if the stream is canceled,
          // but otherwise does not say anything about exceptions that can be thrown from onNext.
          // Note that Blaze redirects System.{out,err} to this output stream, so attempting to call
          // printStackTrace() from here could go into an infinite loop.
          BugReport.sendBugReport(e);
          Thread.currentThread().interrupt();
        }
      }
    }

    @Override
    public void write(int byteAsInt) throws IOException {
      write(new byte[] {(byte) byteAsInt}, 0, 1);
    }
  }

  /**
   * A thread that watches if the PID file changes and shuts down the server immediately if so.
   */
  private class PidFileWatcherThread extends Thread {
    private boolean shuttingDown = false;

    private PidFileWatcherThread() {
      super("pid-file-watcher");
      setDaemon(true);
    }

    // The synchronized block is here so that if the "PID file deleted" timer kicks in during a
    // regular shutdown, they don't race.
    private synchronized void signalShutdown() {
      shuttingDown = true;
    }

    @Override
    public void run() {
      while (true) {
        Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);
        boolean ok = false;
        try {
          String pidFileContents = new String(FileSystemUtils.readContentAsLatin1(pidFile));
          ok = pidFileContents.equals(pidInFile);
        } catch (IOException e) {
          logger.info("Cannot read PID file: " + e.getMessage());
          // Handled by virtue of ok not being set to true
        }

        if (!ok) {
          synchronized (PidFileWatcherThread.this) {
            if (shuttingDown) {
              logger.warning("PID file deleted or overwritten but shutdown is already in progress");
              break;
            }

            shuttingDown = true;
            // Someone overwrote the PID file. Maybe it's another server, so shut down as quickly
            // as possible without even running the shutdown hooks (that would delete it)
            logger.severe("PID file deleted or overwritten, exiting as quickly as possible");
            Runtime.getRuntime().halt(ExitCode.BLAZE_INTERNAL_ERROR.getNumericExitCode());
          }
        }
      }
    }
  }

  // These paths are all relative to the server directory
  private static final String PORT_FILE = "command_port";
  private static final String REQUEST_COOKIE_FILE = "request_cookie";
  private static final String RESPONSE_COOKIE_FILE = "response_cookie";
  private static final String SERVER_INFO_FILE = "server_info.rawproto";

  private static final AtomicBoolean runShutdownHooks = new AtomicBoolean(true);

  private final CommandManager commandManager;
  private final CommandDispatcher dispatcher;
  private final Executor commandExecutorPool;
  private final Clock clock;
  private final Path serverDirectory;
  private final String requestCookie;
  private final String responseCookie;
  private final int maxIdleSeconds;
  private final PidFileWatcherThread pidFileWatcherThread;
  private final Path pidFile;
  private final String pidInFile;
  private final List<Path> filesToDeleteAtExit = new ArrayList<>();
  private final int port;

  private Server server;
  private boolean serving;

  @VisibleForTesting
  GrpcServerImpl(
      CommandDispatcher dispatcher,
      Clock clock,
      int port,
      String requestCookie,
      String responseCookie,
      Path serverDirectory,
      int maxIdleSeconds,
      boolean shutdownOnLowSysMem,
      boolean doIdleServerTasks)
      throws IOException {
    Runtime.getRuntime().addShutdownHook(new Thread(() -> shutdownHook()));

    // server.pid was written in the C++ launcher after fork() but before exec() .
    // The client only accesses the pid file after connecting to the socket
    // which ensures that it gets the correct pid value.
    pidFile = serverDirectory.getRelative("server.pid.txt");
    pidInFile = new String(FileSystemUtils.readContentAsLatin1(pidFile));
    deleteAtExit(pidFile);

    this.dispatcher = dispatcher;
    this.clock = clock;
    this.serverDirectory = serverDirectory;
    this.port = port;
    this.maxIdleSeconds = maxIdleSeconds;
    this.shutdownOnLowSysMem = shutdownOnLowSysMem;
    this.serving = false;

    this.commandExecutorPool =
        Context.currentContextExecutor(
            Executors.newCachedThreadPool(
                new ThreadFactoryBuilder()
                    .setNameFormat("grpc-command-%d")
                    .setDaemon(true)
                    .build()));

    this.requestCookie = requestCookie;
    this.responseCookie = responseCookie;

    pidFileWatcherThread = new PidFileWatcherThread();
    pidFileWatcherThread.start();
    commandManager = new CommandManager(doIdleServerTasks);
  }

  private static String generateCookie(SecureRandom random, int byteCount) {
    byte[] bytes = new byte[byteCount];
    random.nextBytes(bytes);
    StringBuilder result = new StringBuilder();
    for (byte b : bytes) {
      result.append(Integer.toHexString(b + 128));
    }

    return result.toString();
  }

  /**
   * This is called when the server is shut down as a result of a "clean --expunge".
   *
   * <p>In this case, no files should be deleted on shutdown hooks, since clean also deletes the
   * lock file, and there is a small possibility of the following sequence of events:
   *
   * <ol>
   *   <li>Client 1 runs "blaze clean --expunge"
   *   <li>Client 2 runs a command and waits for client 1 to finish
   *   <li>The clean command deletes everything including the lock file
   *   <li>Client 2 starts running and since the output base is empty, starts up a new server, which
   *       creates its own socket and PID files
   *   <li>The server used by client runs its shutdown hooks, deleting the PID files created by the
   *       new server
   * </ol>
   *
   * It also disables the "die when the PID file changes" handler so that it doesn't kill the server
   * while the "clean --expunge" command is running.
   */
  @Override
  public void prepareForAbruptShutdown() {
    disableShutdownHooks();
    pidFileWatcherThread.signalShutdown();
  }

  @Override
  public void interrupt() {
    commandManager.interruptInflightCommands();
  }

  @Override
  public void serve() throws IOException {
    Preconditions.checkState(!serving);

    // For reasons only Apple knows, you cannot bind to IPv4-localhost when you run in a sandbox
    // that only allows loopback traffic, but binding to IPv6-localhost works fine. This would
    // however break on systems that don't support IPv6. So what we'll do is to try to bind to IPv6
    // and if that fails, try again with IPv4.
    InetSocketAddress address = new InetSocketAddress("[::1]", port);
    try {
      server =
          NettyServerBuilder.forAddress(address).addService(this).directExecutor().build().start();
    } catch (IOException e) {
      address = new InetSocketAddress("127.0.0.1", port);
      server =
          NettyServerBuilder.forAddress(address).addService(this).directExecutor().build().start();
    }

    if (maxIdleSeconds > 0) {
      Thread timeoutAndMemoryCheckingThread =
          new Thread(
              new ServerWatcherRunnable(
                  server, maxIdleSeconds, shutdownOnLowSysMem, commandManager));
      timeoutAndMemoryCheckingThread.setName("grpc-timeout-and-memory");
      timeoutAndMemoryCheckingThread.setDaemon(true);
      timeoutAndMemoryCheckingThread.start();
    }
    serving = true;

    writeServerStatusFiles(address);

    try {
      server.awaitTermination();
    } catch (InterruptedException e) {
      // TODO(lberki): Handle SIGINT in a reasonable way
      throw new IllegalStateException(e);
    }
  }

  private void writeServerStatusFiles(InetSocketAddress address) throws IOException {
    String addressString = InetAddresses.toUriString(address.getAddress()) + ":" + server.getPort();
    writeServerFile(PORT_FILE, addressString);
    writeServerFile(REQUEST_COOKIE_FILE, requestCookie);
    writeServerFile(RESPONSE_COOKIE_FILE, responseCookie);

    ServerInfo info =
        ServerInfo.newBuilder()
            .setPid(Integer.parseInt(pidInFile))
            .setAddress(addressString)
            .setRequestCookie(requestCookie)
            .setResponseCookie(responseCookie)
            .build();

    // Write then mv so the user never sees incomplete contents.
    Path serverInfoTmpFile = serverDirectory.getChild(SERVER_INFO_FILE + ".tmp");
    try (OutputStream out = serverInfoTmpFile.getOutputStream()) {
      info.writeTo(out);
    }
    Path serverInfoFile = serverDirectory.getChild(SERVER_INFO_FILE);
    serverInfoTmpFile.renameTo(serverInfoFile);
    deleteAtExit(serverInfoFile);
  }

  private void writeServerFile(String name, String contents) throws IOException {
    Path file = serverDirectory.getChild(name);
    FileSystemUtils.writeContentAsLatin1(file, contents);
    deleteAtExit(file);
  }

  protected void disableShutdownHooks() {
    runShutdownHooks.set(false);
  }

  private void shutdownHook() {
    if (!runShutdownHooks.get()) {
      return;
    }

    List<Path> files;
    synchronized (filesToDeleteAtExit) {
      files = new ArrayList<>(filesToDeleteAtExit);
    }
    for (Path path : files) {
      try {
        path.delete();
      } catch (IOException e) {
        printStack(e);
      }
    }
  }

  /**
   * Schedule the specified file for (attempted) deletion at JVM exit.
   */
  protected void deleteAtExit(final Path path) {
    synchronized (filesToDeleteAtExit) {
      filesToDeleteAtExit.add(path);
    }
  }

  static void printStack(IOException e) {
    /*
     * Hopefully this never happens. It's not very nice to just write this
     * to the user's console, but I'm not sure what better choice we have.
     */
    StringWriter err = new StringWriter();
    PrintWriter printErr = new PrintWriter(err);
    printErr.println("=======[BAZEL SERVER: ENCOUNTERED IO EXCEPTION]=======");
    e.printStackTrace(printErr);
    printErr.println("=====================================================");
    logger.severe(err.toString());
  }

  private void executeCommand(RunRequest request, BlockingStreamObserver<RunResponse> observer) {
    if (!request.getCookie().equals(requestCookie) || request.getClientDescription().isEmpty()) {
      try {
        observer.onNext(
            RunResponse.newBuilder()
                .setExitCode(ExitCode.LOCAL_ENVIRONMENTAL_ERROR.getNumericExitCode())
                .build());
        observer.onCompleted();
      } catch (StatusRuntimeException e) {
        logger.info("Client cancelled command while rejecting it: " + e.getMessage());
      }
      return;
    }

    String commandId;
    BlazeCommandResult result;

    // TODO(b/63925394): This information needs to be passed to the GotOptionsEvent, which does not
    // currently have the explicit startup options. See Improved Command Line Reporting design doc
    // for details.
    // Convert the startup options record to Java strings, source first.
    ImmutableList.Builder<Pair<String, String>> startupOptions = ImmutableList.builder();
    for (StartupOption option : request.getStartupOptionsList()) {
      // UTF-8 won't do because we want to be able to pass arbitrary binary strings.
      // Not that the internals of Bazel handle that correctly, but why not make at least this
      // little part correct?
      startupOptions.add(new Pair<>(
          option.getSource().toString(StandardCharsets.ISO_8859_1),
          option.getOption().toString(StandardCharsets.ISO_8859_1)));
    }

    try (RunningCommand command = commandManager.create()) {
      commandId = command.getId();

      try {
        // Send the client the command id as soon as we know it.
        observer.onNext(
            RunResponse.newBuilder()
                .setCookie(responseCookie)
                .setCommandId(commandId)
                .build());
      } catch (StatusRuntimeException e) {
        logger.info(
            "The client cancelled the command before receiving the command id: " + e.getMessage());
      }

      OutErr rpcOutErr =
          OutErr.create(
              new RpcOutputStream(command.getId(), responseCookie, StreamType.STDOUT, observer),
              new RpcOutputStream(command.getId(), responseCookie, StreamType.STDERR, observer));

      try {
        // UTF-8 won't do because we want to be able to pass arbitrary binary strings.
        // Not that the internals of Bazel handle that correctly, but why not make at least this
        // little part correct?
        ImmutableList<String> args = request.getArgList().stream()
            .map(arg -> arg.toString(StandardCharsets.ISO_8859_1))
            .collect(ImmutableList.toImmutableList());

        InvocationPolicy policy = InvocationPolicyParser.parsePolicy(request.getInvocationPolicy());
        logger.info(BlazeRuntime.getRequestLogString(args));
        result =
            dispatcher.exec(
                policy,
                args,
                rpcOutErr,
                request.getBlockForLock() ? LockingMode.WAIT : LockingMode.ERROR_OUT,
                request.getClientDescription(),
                clock.currentTimeMillis(),
                Optional.of(startupOptions.build()));
      } catch (OptionsParsingException e) {
        rpcOutErr.printErrLn(e.getMessage());
        result = BlazeCommandResult.exitCode(ExitCode.COMMAND_LINE_ERROR);
      }
    } catch (InterruptedException e) {
      result =
          BlazeCommandResult.failureDetail(
              FailureDetailUtil.interrupted(Interrupted.Code.INTERRUPTED_UNSPECIFIED));
      commandId = ""; // The default value, the client will ignore it
    }

    RunResponse.Builder response = RunResponse.newBuilder()
        .setCookie(responseCookie)
        .setCommandId(commandId)
        .setFinished(true)
        .setTerminationExpected(result.shutdown());

    if (result.getExecRequest() != null) {
      response.setExitCode(0);
      response.setExecRequest(result.getExecRequest());
    } else {
      response.setExitCode(result.getExitCode().getNumericExitCode());
      if (result.getFailureDetail() != null) {
        response.setFailureDetail(result.getFailureDetail());
      }
    }

    try {
      observer.onNext(response.build());
      observer.onCompleted();
    } catch (StatusRuntimeException e) {
      logger.info(
          "The client cancelled the command before receiving the command id: " + e.getMessage());
    }

    if (result.shutdown()) {
      server.shutdown();
    }
  }

  @Override
  public void run(final RunRequest request, final StreamObserver<RunResponse> observer) {
    // Switch to our own threads so that onReadyStateHandler can be called (see class-level
    // comment).
    ServerCallStreamObserver<RunResponse> serverCallStreamObserver =
        ((ServerCallStreamObserver<RunResponse>) observer);
    BlockingStreamObserver<RunResponse> blockingStreamObserver =
        new BlockingStreamObserver<>(serverCallStreamObserver);
    commandExecutorPool.execute(() -> executeCommand(request, blockingStreamObserver));
  }

  @Override
  public void ping(PingRequest pingRequest, StreamObserver<PingResponse> streamObserver) {
    try (RunningCommand command = commandManager.create()) {
      PingResponse.Builder response = PingResponse.newBuilder();
      if (pingRequest.getCookie().equals(requestCookie)) {
        response.setCookie(responseCookie);
      }

      streamObserver.onNext(response.build());
      streamObserver.onCompleted();
    }
  }

  @Override
  public void cancel(
      final CancelRequest request, final StreamObserver<CancelResponse> streamObserver) {
    logger.info(String.format("Got CancelRequest for command id %s", request.getCommandId()));
    if (!request.getCookie().equals(requestCookie)) {
      streamObserver.onCompleted();
      return;
    }

    // Actually performing the cancellation can result in some blocking which we don't want
    // to do on the dispatcher thread, instead offload to command pool.
    commandExecutorPool.execute(() -> doCancel(request, streamObserver));
  }

  private void doCancel(CancelRequest request, StreamObserver<CancelResponse> streamObserver) {
    commandManager.doCancel(request);
    try {
      streamObserver.onNext(CancelResponse.newBuilder().setCookie(responseCookie).build());
      streamObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      // There is no one to report the failure to
      logger.info("Client cancelled RPC of cancellation request for " + request.getCommandId());
    }
  }
}