Merge branch '17936-arvput-batch-mode' into main. Closes #17936.

Arvados-DCO-1.1-Signed-off-by: Lucas Di Pentima <lucas.dipentima@curii.com>

diff --combined sdk/python/arvados/commands/put.py
index d8e673bd343d772df9e76fcad6c68facefc55855,daea046a79d959f0e68d435881e8c854da57aaa1..f6f85ba69619ba930cca9efd20d3b4f134f28527
--- 1/sdk/python/arvados/commands/put.py
--- 2/sdk/python/arvados/commands/put.py
+++ b/sdk/python/arvados/commands/put.py
@@@ -215,6 -215,12 +215,12 @@@ Do not print any debug messages to cons
  still be displayed.)
  """)
  
+ run_opts.add_argument('--batch', action='store_true', default=False,
+                       help="""
+ Retries with '--no-resume --no-cache' if cached state contains invalid/expired
+ block signatures.
+ """)
+ 
  _group = run_opts.add_mutually_exclusive_group()
  _group.add_argument('--resume', action='store_true', default=True,
                      help="""
@@@ -438,7 -444,7 +444,7 @@@ class ArvPutUploadJob(object)
      }
  
      def __init__(self, paths, resume=True, use_cache=True, reporter=None,
-                  name=None, owner_uuid=None, api_client=None,
+                  name=None, owner_uuid=None, api_client=None, batch_mode=False,
                   ensure_unique_name=False, num_retries=None,
                   put_threads=None, replication_desired=None, filename=None,
                   update_time=60.0, update_collection=None, storage_classes=None,
@@@ -448,6 -454,7 +454,7 @@@
          self.paths = paths
          self.resume = resume
          self.use_cache = use_cache
+         self.batch_mode = batch_mode
          self.update = False
          self.reporter = reporter
          # This will set to 0 before start counting, if no special files are going
@@@ -908,12 -915,17 +915,17 @@@
                  # No cache file, set empty state
                  self._state = copy.deepcopy(self.EMPTY_STATE)
              if not self._cached_manifest_valid():
-                 raise ResumeCacheInvalidError()
+                 if not self.batch_mode:
+                     raise ResumeCacheInvalidError()
+                 else:
+                     self.logger.info("Invalid signatures on cache file '{}' while being run in 'batch mode' -- continuing anyways.".format(self._cache_file.name))
+                     self.use_cache = False # Don't overwrite preexisting cache file.
+                     self._state = copy.deepcopy(self.EMPTY_STATE)
              # Load the previous manifest so we can check if files were modified remotely.
              self._local_collection = arvados.collection.Collection(
                  self._state['manifest'],
                  replication_desired=self.replication_desired,
 -                storage_classes_desired=(self.storage_classes or ['default']),
 +                storage_classes_desired=self.storage_classes,
                  put_threads=self.put_threads,
                  api_client=self._api_client,
                  num_retries=self.num_retries)
@@@ -1250,6 -1262,7 +1262,7 @@@ def main(arguments=None, stdout=sys.std
          writer = ArvPutUploadJob(paths = args.paths,
                                   resume = args.resume,
                                   use_cache = args.use_cache,
+                                  batch_mode= args.batch,
                                   filename = args.filename,
                                   reporter = reporter,
                                   api_client = api_client,
@@@ -1278,7 -1291,8 +1291,8 @@@
              "         or been created with another Arvados user's credentials.",
              "         Switch user or use one of the following options to restart upload:",
              "         --no-resume to start a new resume cache.",
-             "         --no-cache to disable resume cache."]))
+             "         --no-cache to disable resume cache.",
+             "         --batch to ignore the resume cache if invalid."]))
          sys.exit(1)
      except (CollectionUpdateError, PathDoesNotExistError) as error:
          logger.error("\n".join([