[cdc_stream] [cdc_rsync] Add --forward-port flag (#45)

Adds a flag to set the SSH forwarding port or port range used for
'cdc_stream start-service' and 'cdc_rsync'.

If a single number is passed, e.g. --forward-port 12345, then this
port is used without checking availability of local and remote ports.
If the port is taken, this results in an error when trying to connect.
Note that this restricts the number of connections that stream can
make to one.

If a range is passed, e.g. --forward-port 45000-46000, the tools
search for available ports locally and remotely in that range. This is
more robust, but a bit slower due to the extra overhead.

Optimizes port_manager_win as it was very slow for a large port range.
It's still not optimal, but the time needed to scan 30k ports is
<< 1 seconds now.

Fixes #12

all_files.vcxitems

@@ -34,6 +34,8 @@
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\start_command.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\start_service_command.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\stop_service_command.cc" />
+    <ClCompile Include="$(MSBuildThisFileDirectory)common\port_range_parser.cc" />
+    <ClCompile Include="$(MSBuildThisFileDirectory)common\port_range_parser_test.cc" />
     <ClInclude Include="$(MSBuildThisFileDirectory)cdc_stream\stop_command.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\testing_asset_stream_server.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_fuse_fs\asset.cc" />
@@ -144,6 +146,7 @@
     <ClCompile Include="$(MSBuildThisFileDirectory)metrics\messages_test.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)metrics\metrics.cc" />
     <ClInclude Include="$(MSBuildThisFileDirectory)cdc_stream\stop_service_command.h" />
+    <ClInclude Include="$(MSBuildThisFileDirectory)common\port_range_parser.h" />
   </ItemGroup>
   <ItemGroup>
     <ClInclude Include="$(MSBuildThisFileDirectory)absl_helper\jedec_size_flag.h" />

cdc_rsync/BUILD

@@ -130,6 +130,7 @@ cc_library(
     hdrs = ["params.h"],
     deps = [
         ":cdc_rsync_client",
+        "//common:port_range_parser",
         "@com_github_zstd//:zstd",
         "@com_google_absl//absl/status",
     ],

cdc_rsync/cdc_rsync_client.cc

@@ -44,8 +44,6 @@ constexpr int kExitCodeCouldNotExecute = 126;
 // Bash exit code if binary was not found.
 constexpr int kExitCodeNotFound = 127;
 
-constexpr int kForwardPortFirst = 44450;
-constexpr int kForwardPortLast = 44459;
 constexpr char kCdcServerFilename[] = "cdc_rsync_server";
 constexpr char kRemoteToolsBinDir[] = "~/.cache/cdc-file-transfer/bin/";
 
@@ -104,8 +102,8 @@ CdcRsyncClient::CdcRsyncClient(const Options& options,
                    &process_factory_,
                    /*forward_output_to_log=*/false),
       port_manager_("cdc_rsync_ports_f77bcdfe-368c-4c45-9f01-230c5e7e2132",
-                    kForwardPortFirst, kForwardPortLast, &process_factory_,
-                    &remote_util_),
+                    options.forward_port_first, options.forward_port_last,
+                    &process_factory_, &remote_util_),
       printer_(options.quiet, Util::IsTTY() && !options.json),
       progress_(&printer_, options.verbosity, options.json) {
   if (!options_.ssh_command.empty()) {
@@ -184,19 +182,24 @@ absl::Status CdcRsyncClient::StartServer() {
   std::string component_args = GameletComponent::ToCommandLineArgs(components);
 
   // Find available local and remote ports for port forwarding.
-  absl::StatusOr<int> port_res = port_manager_.ReservePort(
-      /*check_remote=*/false, /*remote_timeout_sec unused*/ 0);
-  constexpr char kErrorMsg[] = "Failed to find available port";
-  if (absl::IsDeadlineExceeded(port_res.status())) {
-    // Server didn't respond in time.
-    return SetTag(WrapStatus(port_res.status(), kErrorMsg),
-                  Tag::kConnectionTimeout);
+  // If only one port is in the given range, try that without checking.
+  int port = options_.forward_port_first;
+  if (options_.forward_port_first < options_.forward_port_last) {
+    absl::StatusOr<int> port_res =
+        port_manager_.ReservePort(options_.connection_timeout_sec);
+    constexpr char kErrorMsg[] = "Failed to find available port";
+    if (absl::IsDeadlineExceeded(port_res.status())) {
+      // Server didn't respond in time.
+      return SetTag(WrapStatus(port_res.status(), kErrorMsg),
+                    Tag::kConnectionTimeout);
+    }
+    if (absl::IsResourceExhausted(port_res.status()))
+      return SetTag(WrapStatus(port_res.status(), kErrorMsg),
+                    Tag::kAddressInUse);
+    if (!port_res.ok())
+      return WrapStatus(port_res.status(), "Failed to find available port");
+    port = *port_res;
   }
-  if (absl::IsResourceExhausted(port_res.status()))
-    return SetTag(WrapStatus(port_res.status(), kErrorMsg), Tag::kAddressInUse);
-  if (!port_res.ok())
-    return WrapStatus(port_res.status(), "Failed to find available port");
-  int port = *port_res;
 
   std::string remote_server_path =
       std::string(kRemoteToolsBinDir) + kCdcServerFilename;

cdc_rsync/cdc_rsync_client.h

@@ -50,6 +50,8 @@ class CdcRsyncClient {
     std::string copy_dest;
     int compress_level = 6;
     int connection_timeout_sec = 10;
+    int forward_port_first = 44450;
+    int forward_port_last = 44459;
     std::string ssh_command;
     std::string scp_command;
     std::string sources_dir;  // Base dir for files loaded for --files-from.

cdc_rsync/params.cc

@@ -20,6 +20,7 @@
 #include "absl/strings/str_format.h"
 #include "absl/strings/str_split.h"
 #include "common/path.h"
+#include "common/port_range_parser.h"
 #include "lib/zstd.h"
 
 namespace cdc_ft {
@@ -45,39 +46,41 @@ For partially matching files only the deltas are transferred.
   cdc_rsync [options] source [source]... [user@]host:destination
 
 Parameters:
-  source                  Local file or directory to be copied
-  user                    Remote SSH user name
-  host                    Remote host or IP address
-  destination             Remote destination directory
+  source                    Local file or directory to be copied
+  user                      Remote SSH user name
+  host                      Remote host or IP address
+  destination               Remote destination directory
 
 Options:
-    --contimeout sec      Gamelet connection timeout in seconds (default: 10)
--q, --quiet               Quiet mode, only print errors
--v, --verbose             Increase output verbosity
-    --json                Print JSON progress
--n, --dry-run             Perform a trial run with no changes made
--r, --recursive           Recurse into directories
-    --delete              Delete extraneous files from destination directory
--z, --compress            Compress file data during the transfer
-    --compress-level num  Explicitly set compression level (default: 6)
--c, --checksum            Skip files based on checksum, not mod-time & size
--W, --whole-file          Always copy files whole,
-                          do not apply delta-transfer algorithm
-    --exclude pattern     Exclude files matching pattern
-    --exclude-from file   Read exclude patterns from file
-    --include pattern     Don't exclude files matching pattern
-    --include-from file   Read include patterns from file
-    --files-from file     Read list of source files from file
--R, --relative            Use relative path names
-    --existing            Skip creating new files on instance
-    --copy-dest dir       Use files from dir as sync base if files are missing
-    --ssh-command         Path and arguments of ssh command to use, e.g.
-                          "C:\path\to\ssh.exe -p 12345 -i id_rsa -oUserKnownHostsFile=known_hosts"
-                          Can also be specified by the CDC_SSH_COMMAND environment variable.
-    --scp-command         Path and arguments of scp command to use, e.g.
-                          "C:\path\to\scp.exe -P 12345 -i id_rsa -oUserKnownHostsFile=known_hosts"
-                          Can also be specified by the CDC_SCP_COMMAND environment variable.
--h  --help                Help for cdc_rsync
+    --contimeout sec        Gamelet connection timeout in seconds (default: 10)
+-q, --quiet                 Quiet mode, only print errors
+-v, --verbose               Increase output verbosity
+    --json                  Print JSON progress
+-n, --dry-run               Perform a trial run with no changes made
+-r, --recursive             Recurse into directories
+    --delete                Delete extraneous files from destination directory
+-z, --compress              Compress file data during the transfer
+    --compress-level <num>  Explicitly set compression level (default: 6)
+-c, --checksum              Skip files based on checksum, not mod-time & size
+-W, --whole-file            Always copy files whole,
+                            do not apply delta-transfer algorithm
+    --exclude pattern       Exclude files matching pattern
+    --exclude-from <file>   Read exclude patterns from file
+    --include pattern       Don't exclude files matching pattern
+    --include-from <file>   Read include patterns from file
+    --files-from <file>     Read list of source files from file
+-R, --relative              Use relative path names
+    --existing              Skip creating new files on instance
+    --copy-dest <dir>       Use files from dir as sync base if files are missing
+    --ssh-command <cmd>     Path and arguments of ssh command to use, e.g.
+                            "C:\path\to\ssh.exe -p 12345 -i id_rsa -oUserKnownHostsFile=known_hosts"
+                            Can also be specified by the CDC_SSH_COMMAND environment variable.
+    --scp-command <cmd>     Path and arguments of scp command to use, e.g.
+                            "C:\path\to\scp.exe -P 12345 -i id_rsa -oUserKnownHostsFile=known_hosts"
+                            Can also be specified by the CDC_SCP_COMMAND environment variable.
+    --forward-port <port>   TCP port or range used for SSH port forwarding (default: 44450-44459).
+                            If a range is specified, searches for available ports (slower).
+-h  --help                  Help for cdc_rsync
 )";
 
 constexpr char kSshCommandEnvVar[] = "CDC_SSH_COMMAND";
@@ -91,15 +94,20 @@ void PopulateFromEnvVars(Parameters* parameters) {
       .IgnoreError();
 }
 
+// Returns false and prints an error if |value| is null or empty.
+bool ValidateValue(const std::string& option_name, const char* value) {
+  if (!value) {
+    PrintError("Option '%s' needs a value", option_name);
+    return false;
+  }
+  return true;
+}
+
 // Handles the --exclude-from and --include-from options.
 OptionResult HandleFilterRuleFile(const std::string& option_name,
                                   const char* path, PathFilter::Rule::Type type,
                                   Parameters* params) {
-  if (!path) {
-    PrintError("Option '%s' needs a value", option_name);
-    return OptionResult::kError;
-  }
-
+  assert(path);
   std::vector<std::string> patterns;
   absl::Status status = path::ReadAllLines(
       path, &patterns,
@@ -188,29 +196,34 @@ OptionResult HandleParameter(const std::string& key, const char* value,
   }
 
   if (key == "include") {
+    if (!ValidateValue(key, value)) return OptionResult::kError;
     params->options.filter.AddRule(PathFilter::Rule::Type::kInclude, value);
     return OptionResult::kConsumedKeyValue;
   }
 
   if (key == "include-from") {
+    if (!ValidateValue(key, value)) return OptionResult::kError;
     return HandleFilterRuleFile(key, value, PathFilter::Rule::Type::kInclude,
                                 params);
   }
 
   if (key == "exclude") {
+    if (!ValidateValue(key, value)) return OptionResult::kError;
     params->options.filter.AddRule(PathFilter::Rule::Type::kExclude, value);
     return OptionResult::kConsumedKeyValue;
   }
 
   if (key == "exclude-from") {
+    if (!ValidateValue(key, value)) return OptionResult::kError;
     return HandleFilterRuleFile(key, value, PathFilter::Rule::Type::kExclude,
                                 params);
   }
 
   if (key == "files-from") {
     // Implies -R.
+    if (!ValidateValue(key, value)) return OptionResult::kError;
     params->options.relative = true;
-    params->files_from = value ? value : std::string();
+    params->files_from = value;
     return OptionResult::kConsumedKeyValue;
   }
 
@@ -225,16 +238,14 @@ OptionResult HandleParameter(const std::string& key, const char* value,
   }
 
   if (key == "compress-level") {
-    if (value) {
-      params->options.compress_level = atoi(value);
-    }
+    if (!ValidateValue(key, value)) return OptionResult::kError;
+    params->options.compress_level = atoi(value);
     return OptionResult::kConsumedKeyValue;
   }
 
   if (key == "contimeout") {
-    if (value) {
-      params->options.connection_timeout_sec = atoi(value);
-    }
+    if (!ValidateValue(key, value)) return OptionResult::kError;
+    params->options.connection_timeout_sec = atoi(value);
     return OptionResult::kConsumedKeyValue;
   }
 
@@ -259,7 +270,8 @@ OptionResult HandleParameter(const std::string& key, const char* value,
   }
 
   if (key == "copy-dest") {
-    params->options.copy_dest = value ? value : std::string();
+    if (!ValidateValue(key, value)) return OptionResult::kError;
+    params->options.copy_dest = value;
     return OptionResult::kConsumedKeyValue;
   }
 
@@ -269,12 +281,27 @@ OptionResult HandleParameter(const std::string& key, const char* value,
   }
 
   if (key == "ssh-command") {
-    params->options.ssh_command = value ? value : std::string();
+    if (!ValidateValue(key, value)) return OptionResult::kError;
+    params->options.ssh_command = value;
     return OptionResult::kConsumedKeyValue;
   }
 
   if (key == "scp-command") {
-    params->options.scp_command = value ? value : std::string();
+    if (!ValidateValue(key, value)) return OptionResult::kError;
+    params->options.scp_command = value;
+    return OptionResult::kConsumedKeyValue;
+  }
+
+  if (key == "forward-port") {
+    if (!ValidateValue(key, value)) return OptionResult::kError;
+    uint16_t first, last;
+    if (!port_range::Parse(value, &first, &last)) {
+      PrintError("Failed to parse %s=%s, expected <port> or <port1>-<port2>",
+                 key, value);
+      return OptionResult::kError;
+    }
+    params->options.forward_port_first = first;
+    params->options.forward_port_last = last;
     return OptionResult::kConsumedKeyValue;
   }
 
@@ -355,11 +382,7 @@ bool CheckOptionResult(OptionResult result, const std::string& name,
       return true;
 
     case OptionResult::kConsumedKeyValue:
-      if (!value) {
-        PrintError("Option '%s' needs a value", name);
-        return false;
-      }
-      return true;
+      return ValidateValue(name, value);
 
     case OptionResult::kError:
       // Error message was already printed.

cdc_rsync/params_test.cc

@@ -536,6 +536,40 @@ TEST_F(ParamsTest, IncludeExcludeMixed_ProperOrder) {
   ExpectNoError();
 }
 
+TEST_F(ParamsTest, ForwardPort_Single) {
+  const char* argv[] = {"cdc_rsync.exe", "--forward-port=65535", kSrc,
+                        kUserHostDst, NULL};
+  EXPECT_TRUE(Parse(static_cast<int>(std::size(argv)) - 1, argv, &parameters_));
+  EXPECT_EQ(parameters_.options.forward_port_first, 65535);
+  EXPECT_EQ(parameters_.options.forward_port_last, 65535);
+  ExpectNoError();
+}
+
+TEST_F(ParamsTest, ForwardPort_Range) {
+  const char* argv[] = {
+      "cdc_rsync.exe", "--forward-port", "1-2", kSrc, kUserHostDst, NULL};
+  EXPECT_TRUE(Parse(static_cast<int>(std::size(argv)) - 1, argv, &parameters_));
+  EXPECT_EQ(parameters_.options.forward_port_first, 1);
+  EXPECT_EQ(parameters_.options.forward_port_last, 2);
+  ExpectNoError();
+}
+
+TEST_F(ParamsTest, ForwardPort_NoValue) {
+  const char* argv[] = {"cdc_rsync.exe", "--forward-port=", kSrc, kUserHostDst,
+                        NULL};
+  EXPECT_FALSE(
+      Parse(static_cast<int>(std::size(argv)) - 1, argv, &parameters_));
+  ExpectError(NeedsValueError("forward-port"));
+}
+
+TEST_F(ParamsTest, ForwardPort_BadValueTooSmall) {
+  const char* argv[] = {"cdc_rsync.exe", "--forward-port=0", kSrc, kUserHostDst,
+                        NULL};
+  EXPECT_FALSE(
+      Parse(static_cast<int>(std::size(argv)) - 1, argv, &parameters_));
+  ExpectError("Failed to parse");
+}
+
 }  // namespace
 }  // namespace params
 }  // namespace cdc_ft

cdc_stream/BUILD

@@ -24,6 +24,7 @@ cc_library(
     hdrs = ["base_command.h"],
     deps = [
         "//absl_helper:jedec_size_flag",
+        "//common:port_range_parser",
         "@com_github_lyra//:lyra",
         "@com_google_absl//absl/status",
         "@com_google_absl//absl/strings:str_format",