Exportar arquivos para Cloudflare R2 com Scala e o AWS SDK

Kevin van Zonneveld

Cofundador · Amsterdã, Países Baixos ·

Use o AWS SDK for Java v2 para exportar arquivos locais finalizados de uma aplicação Scala para um bucket privado do Cloudflare R2. Este exemplo oferece um programa de linha de comando com modos de upload síncrono, assíncrono, em streaming com FS2 e paralelo, além de uma forma de verificar os bytes armazenados.

Por que escolher o Cloudflare R2?

A API compatível com S3 do R2 permite que o código Scala use a assinatura de requisições e os clientes HTTP do SDK Java. O exemplo usa um bucket sem restrição de jurisdição no endpoint padrão da conta. Buckets específicos de uma jurisdição precisam do endpoint correspondente.

Configurar seu projeto Scala

Esta receita se destina a aplicações existentes em Scala 2.13 que usam sbt 1.x. As versões fixadas e testadas são Scala 2.13.18, sbt 1.13.0, Cats Effect 3.7.1, FS2 3.14.0 e AWS SDK 2.56.0. Scala 3 e sbt 2 são opções de migração separadas; essas versões mantêm o exemplo na família de builds Scala 2.13.

Use Bash, cURL e uma instalação do OpenJDK 21 com suporte ativo, com java no seu PATH. A versão de correção do JDK testada é 21.0.12.1; use a atualização de segurança atual do seu distribuidor do JDK na linha 21. Você não precisa de um compilador Scala nem do comando sbt instalados separadamente. Execute o seguinte em um diretório com permissão de escrita para criar um projeto; o subshell mantém seu diretório atual inalterado:

(
  set -eu
  command -v java >/dev/null
  command -v curl >/dev/null
  java -version
  mkdir r2-scala
  cd r2-scala
  mkdir -p project src/main/scala
  curl -fsSLo sbt-launch.jar https://repo.maven.apache.org/maven2/org/scala-sbt/sbt-launch/1.13.0/sbt-launch-1.13.0.jar
)

Se a criação falhar porque r2-scala existe, escolha outro diretório pai. Se o download falhar, mantenha o projeto e tente executar apenas o comando cURL novamente dentro dele. Salve os arquivos a seguir em r2-scala. Use project/build.properties para fixar a versão do inicializador:

sbt.version=1.13.0

Salve build.sbt:

scalaVersion := "2.13.18"

libraryDependencies ++= Seq(
  "software.amazon.awssdk" % "s3" % "2.56.0",
  "software.amazon.awssdk" % "netty-nio-client" % "2.56.0",
  "software.amazon.awssdk" % "apache-client" % "2.56.0",
  "org.typelevel" %% "cats-effect" % "3.7.1",
  "co.fs2" %% "fs2-io" % "3.14.0",
  "co.fs2" %% "fs2-reactive-streams" % "3.14.0"
)

Compile / run / fork := true

Configurar o cliente do AWS SDK

Crie um bucket e credenciais S3 do R2 restritas a esse bucket, com permissão de leitura e escrita de objetos. Forneça R2_ACCOUNT_ID, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY e R2_BUCKET pelo ambiente do processo antes de executar o programa. Use um gerenciador de segredos ou a configuração de ambiente do seu terminal; não coloque credenciais no arquivo Scala nem nos argumentos do comando. O bucket permanece privado e já deve existir.

Os clientes usam a região auto, endereçamento no estilo de caminho e codificação em blocos desativada, conforme a configuração Java da Cloudflare. WHEN_REQUIRED desativa as somas de verificação opcionais das requisições do SDK. O programa seleciona Apache para HTTP síncrono e Netty para HTTP assíncrono, e fecha cada cliente com Resource do Cats Effect.

Upload síncrono básico

Escolha sync para um PUT bloqueante simples. Ele lê o arquivo por meio de RequestBody.fromFile no pool de threads para operações bloqueantes do Cats Effect. O cancelamento aguarda a chamada bloqueante retornar ou atingir o tempo limite; escolha async ou fs2 quando precisar cancelar o future subjacente.

Upload assíncrono com Cats Effect

Salve este programa completo como src/main/scala/R2Upload.scala. async usa o corpo de requisição baseado em arquivo do SDK; fs2 fornece um publicador Reactive Streams. Ambos aguardam por meio da ponte para CompletableFuture do Cats Effect, que solicita o cancelamento do future quando sua fibra é cancelada.

import cats.effect.{ExitCode, IO, IOApp, Resource}
import cats.effect.implicits._
import cats.syntax.all._
import fs2.interop.reactivestreams._
import fs2.io.file.{Files, Flags, Path}
import java.io.InputStream
import java.net.URI
import java.nio.ByteBuffer
import java.security.MessageDigest
import java.time.Duration
import java.util.Optional
import org.reactivestreams.Subscriber
import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
import software.amazon.awssdk.core.async.AsyncRequestBody
import software.amazon.awssdk.core.checksums.RequestChecksumCalculation
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
import software.amazon.awssdk.core.sync.RequestBody
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.retries.StandardRetryStrategy
import software.amazon.awssdk.services.s3.{S3AsyncClient, S3Client, S3Configuration}
import software.amazon.awssdk.services.s3.model.{GetObjectRequest, PutObjectRequest, PutObjectResponse, S3Exception}

object R2Upload extends IOApp {
  final case class Config(endpoint: URI, accessKey: String, secretKey: String, bucket: String)
  final case class Task(path: Path, key: String)

  private val service = S3Configuration.builder()
    .pathStyleAccessEnabled(true)
    .chunkedEncodingEnabled(false)
    .build()

  private def overrides: ClientOverrideConfiguration =
    ClientOverrideConfiguration.builder()
      .apiCallTimeout(Duration.ofMinutes(2))
      .apiCallAttemptTimeout(Duration.ofSeconds(45))
      .retryStrategy(StandardRetryStrategy.builder().maxAttempts(3).build())
      .build()

  private def credentials(config: Config): StaticCredentialsProvider =
    StaticCredentialsProvider.create(
      AwsBasicCredentials.create(config.accessKey, config.secretKey)
    )

  def asyncClient(config: Config): Resource[IO, S3AsyncClient] =
    Resource.fromAutoCloseable(IO.blocking {
      S3AsyncClient.builder()
        .httpClientBuilder(NettyNioAsyncHttpClient.builder()
          .connectionTimeout(Duration.ofSeconds(10))
          .readTimeout(Duration.ofSeconds(30))
          .writeTimeout(Duration.ofSeconds(30)))
        .endpointOverride(config.endpoint)
        .credentialsProvider(credentials(config))
        .region(Region.of("auto"))
        .serviceConfiguration(service)
        .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
        .overrideConfiguration(overrides)
        .build()
    })

  def syncClient(config: Config): Resource[IO, S3Client] =
    Resource.fromAutoCloseable(IO.blocking {
      S3Client.builder()
        .httpClientBuilder(ApacheHttpClient.builder()
          .connectionTimeout(Duration.ofSeconds(10))
          .socketTimeout(Duration.ofSeconds(30)))
        .endpointOverride(config.endpoint)
        .credentialsProvider(credentials(config))
        .region(Region.of("auto"))
        .serviceConfiguration(service)
        .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
        .overrideConfiguration(overrides)
        .build()
    })

  private def request(bucket: String, task: Task, size: Long): PutObjectRequest = {
    require(size >= 0 && size <= 5L * 1024 * 1024 * 1024,
      "Use multipart upload for files larger than 5 GiB")
    PutObjectRequest.builder()
      .bucket(bucket).key(task.key)
      .contentType("application/octet-stream")
      .contentLength(size)
      .build()
  }

  def uploadSync(client: S3Client, bucket: String, task: Task): IO[PutObjectResponse] =
    Files[IO].size(task.path).flatMap { size =>
      IO.blocking(client.putObject(
        request(bucket, task, size), RequestBody.fromFile(task.path.toNioPath)
      ))
    }

  def uploadAsync(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
    Files[IO].size(task.path).flatMap { size =>
      IO.fromCompletableFuture(IO {
        client.putObject(request(bucket, task, size),
          AsyncRequestBody.fromFile(task.path.toNioPath))
      })
    }

  def uploadStream(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
    Files[IO].size(task.path).flatMap { size =>
      val buffers = Files[IO].readAll(task.path, 64 * 1024, Flags.Read)
        .chunks.map(chunk => ByteBuffer.wrap(chunk.toArray))

      buffers.toUnicastPublisher.use { publisher =>
        val body = new AsyncRequestBody {
          override def contentLength(): Optional[java.lang.Long] =
            Optional.of(java.lang.Long.valueOf(size))

          override def subscribe(subscriber: Subscriber[_ >: ByteBuffer]): Unit =
            publisher.subscribe(subscriber)
        }
        IO.fromCompletableFuture(IO {
          client.putObject(request(bucket, task, size), body)
        })
      }
    }

  def uploadMany(client: S3AsyncClient, bucket: String,
      tasks: List[Task]): IO[List[PutObjectResponse]] =
    tasks.parTraverseN(4)(task => uploadAsync(client, bucket, task))

  private def digest(input: InputStream): Array[Byte] = {
    val hash = MessageDigest.getInstance("SHA-256")
    val buffer = new Array[Byte](64 * 1024)
    var count = input.read(buffer)
    while (count != -1) {
      hash.update(buffer, 0, count)
      count = input.read(buffer)
    }
    hash.digest()
  }

  def verify(client: S3Client, bucket: String, task: Task): IO[Unit] = {
    val local = Resource.fromAutoCloseable(IO.blocking(
      java.nio.file.Files.newInputStream(task.path.toNioPath)))
    val remote = Resource.fromAutoCloseable(IO.blocking(client.getObject(
      GetObjectRequest.builder().bucket(bucket).key(task.key).build())))
    (local, remote).tupled.use { case (source, stored) =>
      IO.blocking {
        require(MessageDigest.isEqual(digest(source), digest(stored)), "Stored bytes differ")
      }
    }
  }

  override def run(args: List[String]): IO[ExitCode] = {
    val operation = IO {
      require(args.length >= 3 && args.length % 2 == 1,
        "Usage: R2Upload sync|async|fs2|parallel|verify file key [file key ...]")
      val mode = args.head
      require(Set("sync", "async", "fs2", "parallel", "verify").contains(mode), "Unknown mode")
      require(mode == "parallel" || args.length == 3, "Only parallel accepts multiple files")
      val account = sys.env("R2_ACCOUNT_ID")
      require(account.matches("[a-fA-F0-9]{32}"), "Invalid R2 account ID")
      val config = Config(
        URI.create(s"https://$account.r2.cloudflarestorage.com"),
        sys.env("R2_ACCESS_KEY_ID"), sys.env("R2_SECRET_ACCESS_KEY"), sys.env("R2_BUCKET")
      )
      val tasks = args.tail.grouped(2).map(pair => Task(Path(pair.head), pair(1))).toList
      require(tasks.forall(task =>
        task.path.toString.matches("[A-Za-z0-9_./-]+") &&
        task.key.matches("[A-Za-z0-9_./-]+") && !task.key.startsWith("-")),
        "Use ASCII paths and keys containing only letters, digits, _, ., /, and -")
      require(tasks.map(_.key).distinct.size == tasks.size, "Use distinct destination keys")
      (mode, config, tasks)
    }.flatMap { case (mode, config, tasks) =>
      val upload = if (mode == "verify")
        syncClient(config).use(client => verify(client, config.bucket, tasks.head))
      else if (mode == "sync")
        syncClient(config).use(client => uploadSync(client, config.bucket, tasks.head).void)
      else
        asyncClient(config).use { client =>
          mode match {
            case "fs2" => uploadStream(client, config.bucket, tasks.head).void
            case "parallel" => uploadMany(client, config.bucket, tasks).void
            case _ => uploadAsync(client, config.bucket, tasks.head).void
          }
        }
      upload *> IO.println(if (mode == "verify") "Stored bytes match." else "Upload completed.")
    }

    operation.as(ExitCode.Success).handleErrorWith {
      case error: S3Exception =>
        IO.println(s"Upload failed (${error.getClass.getSimpleName}, HTTP ${error.statusCode()}).").as(ExitCode.Error)
      case error =>
        IO.println(s"Upload failed (${error.getClass.getSimpleName}). Check arguments, environment, file access, and network.")
          .as(ExitCode.Error)
    }
  }
}

Abra um terminal em r2-scala. Estes comandos criam uma pequena entrada, compilam o programa e fazem upload dela. A primeira compilação baixa as dependências. Os argumentos de arquivo e chave neste inicializador se limitam a letras ASCII, dígitos, _, ., / e -; use ./ antes de um nome local que comece com hífen. As chaves de destino não devem começar com - e devem ser distintas dentro de um lote. Mantenha os arquivos de origem inalterados até que todas as tentativas terminem.

(
  set -eu
  printf 'Hello, R2!\n' > hello.txt
  java -Dsbt.server.autostart=false -jar sbt-launch.jar compile
  java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload async ./hello.txt uploads/hello.txt"
)

Substitua async por sync ou fs2 para usar os outros modos de arquivo único. Um PUT bem-sucedido substitui qualquer objeto na mesma chave, inclusive quando a entrada está vazia. Use uma nova chave se precisar manter o objeto anterior.

Após o sucesso, leia novamente o objeto privado e compare os hashes SHA-256 do conteúdo completo local e do conteúdo completo armazenado:

java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload verify ./hello.txt uploads/hello.txt"

O verificador imprime Stored bytes match. somente quando os dois hashes coincidem e retorna um status diferente de zero em caso de divergência ou falha de leitura. Um PUT concluído, por si só, confirma a aceitação pelo serviço; execute a comparação antes de excluir um original importante. A verificação também exige que a origem permaneça inalterada.

Tratar respostas e erros

Upload completed. aparece somente após a conclusão de todos os uploads selecionados e do escopo do cliente. Erros do serviço retornam um código de saída diferente de zero e o status HTTP. Erros de configuração local, arquivo e rede também retornam um código de saída diferente de zero, sem imprimir credenciais nem respostas brutas do SDK.

Um upload cancelado não passa a ser uma conclusão bem-sucedida. O cancelamento ou a perda de uma resposta ainda pode deixar um objeto efetivado no destino: inspecione-o antes de tentar novamente. Este programa não reverte objetos nem exclui dados remotos em caso de falha.

Lógica de novas tentativas com espera exponencial

A estratégia padrão de novas tentativas do SDK trata falhas que permitem uma nova tentativa, com no máximo três tentativas por PUT. O tempo limite da chamada à API é de dois minutos, incluindo novas tentativas; o tempo limite de cada tentativa é de 45 segundos. O estabelecimento da conexão tem um limite de 10 segundos. Esses limites se aplicam às requisições, não são um prazo de dois minutos para um lote paralelo inteiro. Os limites de leitura de 30 segundos dos transportes ficam abaixo do tempo limite de cada tentativa; o Netty também tem um limite de escrita de 30 segundos.

Uma transferência lenta pode atingir o tempo limite mesmo abaixo do limite de tamanho de arquivo. Verifique o destino e ajuste os limites para sua conexão antes de tentar novamente com o arquivo inalterado. Não adicione um loop irrestrito de novas tentativas. O publicador FS2 pode executar novamente seu fluxo de arquivo para uma nova assinatura, por isso também precisa de uma origem estável que possa ser lida novamente.

Fazer upload de muitos arquivos em paralelo

O modo parallel executa no máximo quatro efeitos de upload ao mesmo tempo. Se um falhar, ele cancela os efeitos irmãos e retorna um status de falha. Os objetos concluídos permanecem; um lote não é uma transação. Após criar um segundo arquivo local chamado report.pdf, execute:

java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload parallel ./hello.txt uploads/hello.txt ./report.pdf uploads/report.pdf"

O limite abrange os uploads ativos, não a lista de argumentos em memória nem todos os buffers internos do Netty e do SDK. Ele não divide um arquivo em partes paralelas.

Transmitir arquivos grandes em streaming com fs2

fs2 lê blocos de 64 KiB e os expõe como elementos ByteBuffer conforme a demanda do consumidor. O recurso do publicador permanece aberto até a conclusão, a falha ou o cancelamento. Isso evita carregar o arquivo inteiro em um array de bytes; não é um limite fixo para a memória total da JVM ou do cliente HTTP.

Todos os modos de upload usam PUTs únicos e rejeitam arquivos maiores que 5 GiB antes de enviar uma requisição. O streaming não adiciona upload multipart nem retomada. Para arquivos maiores ou partes que possam ser retomadas, use uploads multipart do R2.

Boas práticas de segurança

Use credenciais restritas ao bucket e escolha as chaves de destino com cuidado. Este programa Scala executado no servidor não precisa de configuração CORS no navegador nem de uma URL pública do bucket. Para arquivos fornecidos por usuários, conclua a validação antes de exportá-los para um local cujo conteúdo sua aplicação disponibiliza.

Erros comuns e suas correções

Resultado	O que verificar
Falha HTTP	Verifique o endpoint da conta, o nome do bucket e as permissões de objeto do token. Um status, por si só, não identifica qual configuração está incorreta.
Falha local	Verifique as quatro variáveis de ambiente, as restrições dos argumentos, se os arquivos podem ser lidos e o limite de 5 GiB.
Tempo limite excedido ou falha de conexão	Verifique a conectividade e os limites das requisições, depois inspecione o destino antes de tentar novamente.
Lote paralelo parcial	Inspecione cada chave solicitada; os uploads irmãos bem-sucedidos permanecem, mesmo que o comando como um todo tenha falhado.

Conclusão

Use o corpo de requisição baseado em arquivo do SDK para exportações comuns e FS2 quando seu pipeline de streaming existente precisar fornecer os bytes. Se a exportação ocorrer após um Job de processamento da Transloadit, o Robot 🤖 /cloudflare/store pode gravar os resultados no R2 como alternativa à execução deste programa.