public Flux<CopyResult> copyToBackup(final AuthenticatedBackupUser backupUser, List<CopyParameters> toCopy) {
    checkBackupLevel(backupUser, BackupLevel.MEDIA);

    return Mono
        // Figure out how many objects we're allowed to copy, updating the quota usage for the amount we are allowed
        .fromFuture(enforceQuota(backupUser, toCopy))

        // Copy the ones we have enough quota to hold
        .flatMapMany(quotaResult -> Flux.concat(

            // These fit in our remaining quota, so perform the copy. If the copy fails, our estimated quota usage may not
            // be exact since we already updated our usage. We make a best-effort attempt to undo the usage update if we
            // know that the copied failed for sure though.
            Flux.fromIterable(quotaResult.requestsToCopy()).flatMapSequential(
                copyParams -> copyToBackup(backupUser, copyParams)
                    .flatMap(copyResult -> switch (copyResult.outcome()) {
                      case SUCCESS -> Mono.just(copyResult);
                      case SOURCE_WRONG_LENGTH, SOURCE_NOT_FOUND, OUT_OF_QUOTA -> Mono
                          .fromFuture(this.backupsDb.trackMedia(backupUser, -1, -copyParams.destinationObjectSize()))
                          .thenReturn(copyResult);
                    }),
                COPY_CONCURRENCY),

            // There wasn't enough quota remaining to perform these copies
            Flux.fromIterable(quotaResult.requestsToReject())
                .map(arg -> new CopyResult(CopyResult.Outcome.OUT_OF_QUOTA, arg.destinationMediaId(), null))));
  }
@Test
  public void copyPartialSuccess() {
    final AuthenticatedBackupUser backupUser = backupUser(TestRandomUtil.nextBytes(16), BackupLevel.MEDIA);
    final List<CopyParameters> toCopy = List.of(
        new CopyParameters(3, "success", 100, COPY_ENCRYPTION_PARAM, TestRandomUtil.nextBytes(15)),
        new CopyParameters(3, "missing", 200, COPY_ENCRYPTION_PARAM, TestRandomUtil.nextBytes(15)),
        new CopyParameters(3, "badlength", 300, COPY_ENCRYPTION_PARAM, TestRandomUtil.nextBytes(15)));

    when(tusCredentialGenerator.generateUpload(any()))
        .thenReturn(new BackupUploadDescriptor(3, "", Collections.emptyMap(), ""));
    when(remoteStorageManager.copy(eq(3), eq("success"), eq(100), any(), any()))
        .thenReturn(CompletableFuture.completedFuture(null));
    when(remoteStorageManager.copy(eq(3), eq("missing"), eq(200), any(), any()))
        .thenReturn(CompletableFuture.failedFuture(new SourceObjectNotFoundException()));
    when(remoteStorageManager.copy(eq(3), eq("badlength"), eq(300), any(), any()))
        .thenReturn(CompletableFuture.failedFuture(new InvalidLengthException("")));

    final List<CopyResult> results = backupManager.copyToBackup(backupUser, toCopy)
        .collectList().block();

    assertThat(results.get(0).outcome()).isEqualTo(CopyResult.Outcome.SUCCESS);
    assertThat(results.get(1).outcome()).isEqualTo(CopyResult.Outcome.SOURCE_NOT_FOUND);
    assertThat(results.get(2).outcome()).isEqualTo(CopyResult.Outcome.SOURCE_WRONG_LENGTH);

    // usage should be rolled back after a known copy failure
    final Map<String, AttributeValue> backup = getBackupItem(backupUser);
    assertThat(AttributeValues.getLong(backup, BackupsDb.ATTR_MEDIA_BYTES_USED, -1L))
        .isEqualTo(toCopy.get(0).destinationObjectSize());
    assertThat(AttributeValues.getLong(backup, BackupsDb.ATTR_MEDIA_COUNT, -1L)).isEqualTo(1L);
  }