GcpGcs

A Google Cloud Storage client for Elixir, built on the JSON API over Finch. Bucket and object management, simple and resumable uploads, and streamed downloads — with structured errors, token caching, and telemetry.

CI Hex.pm

Features

Installation

def deps do
  [
    {:gcp_gcs, "~> 0.3.0"},
    {:goth, "~> 1.4"}  # optional, service-account auth
  ]
end

Quick start

Authenticate

# Development: use your own credentials
gcloud auth application-default login

# Production: a service account file (consumed via Goth)
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

Use it

project = "my-project"

# Create a bucket
{:ok, _bucket} = GcpGcs.create_bucket(project, "my-bucket", location: "US")

# Upload + download a small object
{:ok, _object} = GcpGcs.put_object("my-bucket", "hello.txt", "Hello!", content_type: "text/plain")
{:ok, "Hello!"} = GcpGcs.download("my-bucket", "hello.txt")

# Object metadata
{:ok, object} = GcpGcs.get_object("my-bucket", "hello.txt")
object["size"]         #=> "6"
object["contentType"]  #=> "text/plain"

# List objects (with pagination info)
{:ok, %{items: items, next_page_token: token}} = GcpGcs.list_objects("my-bucket")

# ...or every object, fetched lazily page by page (raises GcpGcs.Error on failure)
"my-bucket" |> GcpGcs.stream_objects!(page_size: 1000) |> Stream.map(& &1["name"]) |> Enum.take(5)

Every call returns {:ok, result} / :ok, or {:error, %GcpGcs.Error{}}; only the lazy stream_objects!/2 and stream_buckets!/2 raise. Resources are decoded JSON — plain maps with string keys; list results are atom-keyed wrappers (%{items: ..., next_page_token: ...}) around them. Unknown options are rejected with :validation_error.

Uploads

# Simple (in-memory) upload
{:ok, _} = GcpGcs.put_object("my-bucket", "notes.txt", "some text")

# Resumable upload of a local file, in 8 MiB chunks read from disk
{:ok, _} = GcpGcs.upload_file("my-bucket", "video.mp4", "/path/to/video.mp4")

# Resumable upload from any enumerable of binaries
chunks = Stream.repeatedly(fn -> :crypto.strong_rand_bytes(1_000_000) end) |> Stream.take(10)
{:ok, _} = GcpGcs.upload_stream("my-bucket", "random.bin", chunks)

# In-memory data with object metadata (goes through a resumable session)
{:ok, _} = GcpGcs.put_object("my-bucket", "a.json", "{}", attrs: %{"cacheControl" => "no-cache"})

# Let the library pick the strategy
{:ok, _} = GcpGcs.upload("my-bucket", "a.txt", "inline data")
{:ok, _} = GcpGcs.upload("my-bucket", "b.bin", {:file, "/path/to/b.bin"})

Content type is guessed from the object's extension for upload_file/4; pass content_type: to override. Resumable chunk size defaults to 8 MiB and can be tuned with chunk_size: (rounded to a 256 KiB multiple).

Resumable uploads survive transient failures (5xx, 429, dropped connections): they ask the session how much it persisted and resume from there. upload_file/4 can resume from any offset; upload_stream/4 holds at most one chunk and returns :data_loss if the server lost more. The final request carries an MD5 X-Goog-Hash of the whole object, so GCS rejects corrupted uploads (checksum: false to disable). After a session exists, errors carry details: %{session_uri: uri, cause: ...} for manual resumption.

Low-level resumable control

{:ok, session} = GcpGcs.start_resumable_upload("my-bucket", "big.bin", content_type: "application/octet-stream")
{:resume, 262144} = GcpGcs.upload_chunk(session, chunk1, offset: 0, total: nil)
{:ok, _object}    = GcpGcs.upload_chunk(session, last_chunk, offset: 262144, total: 300000)

{:resume, persisted} reports how many bytes the session holds; if it is less than what you sent, resend from persisted.

Session URIs are bearer capabilities. Anyone holding one can write the object until the session expires (one week), with no further credentials, so:

For browser uploads, take :origin only from an allow-list of your own origins, never from the request's Origin header as-is: GCS CORS-enables the session for exactly that origin.

Downloads

# Whole object into memory (peak about 2x its size while the body is joined;
# use download_to_file or download_stream for large objects)
{:ok, bytes} = GcpGcs.download("my-bucket", "report.pdf")

# Stream straight to disk (constant memory)
:ok = GcpGcs.download_to_file("my-bucket", "report.pdf", "/tmp/report.pdf")

# Fold the body through your own reducer; return {:halt, acc} to stop early
{:ok, byte_count} =
  GcpGcs.download_stream("my-bucket", "big.bin", 0, fn
    {:data, chunk}, n -> n + byte_size(chunk)
    _other, n -> n
  end)

# Byte ranges (inclusive): a 206 Partial Content is a success
{:ok, first_kib} = GcpGcs.download("my-bucket", "big.bin", range: 0..1023)

download_to_file/4 writes to path <> ".part" and renames it into place when complete, so a partial file never appears at path.

Listing a "directory"

{:ok, %{items: files, prefixes: subdirs}} =
  GcpGcs.list_objects("my-bucket", prefix: "logs/2026/", delimiter: "/")

Copy, compose, rewrite, move

# Copy (same-location)
{:ok, _} = GcpGcs.copy_object("src-bucket", "a.txt", "dst-bucket", "a-copy.txt")

# Compose (concatenate up to 32 objects in one bucket)
{:ok, _} = GcpGcs.compose_object("my-bucket", "combined.log", ["part-1", "part-2", "part-3"])

# Rewrite (large / cross-location / storage-class change — follows rewrite tokens)
{:ok, _} = GcpGcs.rewrite_object("src", "big.bin", "dst", "big.bin")

# Move/rename (buckets with hierarchical namespace)
{:ok, _} = GcpGcs.move_object("my-bucket", "old/name.txt", "new/name.txt")

Error handling

case GcpGcs.get_object("my-bucket", "missing.txt") do
  {:ok, object} ->
    object

  {:error, %GcpGcs.Error{code: :not_found}} ->
    :missing

  {:error, %GcpGcs.Error{code: :unauthenticated}} ->
    {:error, :auth}

  {:error, %GcpGcs.Error{} = err} ->
    raise "GCS error: #{err}"
end

code is a stable atom (t:GcpGcs.Error.code/0 lists them all). HTTP statuses map to :not_found, :already_exists, :permission_denied, :unauthenticated, :invalid_argument, :failed_precondition, :resource_exhausted, :unavailable, …; client-side failures are :validation_error (nothing was sent), :connection_error, :protocol_error (the server broke the resumable/rewrite protocol), :data_loss and :io_error. The original decoded body / exception is preserved in details.

Configuration

Production

No configuration required — requests go to storage.googleapis.com.

# Optional: authenticate through a Goth instance instead of the gcloud CLI
config :gcp_gcs, :goth, MyApp.Goth
# Optional: tune the Finch pool (defaults: one shard of 50 connections)
config :gcp_gcs, :finch, pool_size: 50, pool_count: 1

# Optional: HTTP/2 (opt-in; default HTTP/1)
config :gcp_gcs, :finch, protocols: [:http2]

# Optional: bring your own Finch instance
config :gcp_gcs, :start_finch, false
config :gcp_gcs, :finch_name, MyApp.Finch

# Optional: default request timeout (ms)
config :gcp_gcs, :default_timeout, 30_000

# Optional: max wait for a pooled connection (ms); exceeding it returns
# {:error, %GcpGcs.Error{code: :resource_exhausted}}
config :gcp_gcs, :pool_timeout, 5_000

# Optional: retries for idempotent requests (and resumable-upload recovery)
config :gcp_gcs, :retry, max_attempts: 4, base_delay: 200, max_delay: 10_000

Pool sizing: every in-flight upload or download holds one connection for its whole duration, so pool_size * pool_count is your transfer concurrency and must cover the peak number of concurrent transfers plus metadata calls; otherwise callers wait up to :pool_timeout and then get :resource_exhausted. Raise pool_size first: Finch picks a shard at random per request, so several shards start queueing while one pool of the same total size still has free connections. HTTP/2 multiplexes requests over fewer connections; it relies on the mint ≥ 1.11 / hpax ≥ 1.1 fixes pulled in by finch ≥ 0.24.

TLS: HTTPS connections verify the server certificate against the OS trust store, which Mint loads at connect time (CAStore's bundle if the OS store is missing and castore is a dependency). The store is loaded once at application start, so a host without one fails at boot. If you override config :gcp_gcs, :ssl_opts, point at a bundle with cacertfile: rather than embedding a decoded cacerts: list: the options are copied with the connection on every pool checkout, so a large list costs hundreds of microseconds per request.

Per call, every function accepts :timeout, :pool_timeout, :retry (false or overrides), :finch, :goth and :base_url (a per-call endpoint, which keeps authentication on). Metadata calls also accept :request_timeout, an overall deadline.

Development / test (emulator)

config :gcp_gcs, :emulator,
  scheme: "http",
  host: "localhost",
  port: 4443

STORAGE_EMULATOR_HOST (used by Google's own client libraries) is also honored; set config :gcp_gcs, :emulator_env, false to ignore it (e.g. in production releases). When an emulator is configured, authentication is skipped, and an emulator on a non-loopback host is logged as an error at boot. The endpoint is resolved once at application start.

Authentication

GcpGcs.Auth obtains OAuth2 access tokens:

  1. Goth — a per-call goth: option, or config :gcp_gcs, :goth, MyApp.Goth (Goth caches and refreshes its own tokens)
  2. gcloud CLI — gcloud auth application-default print-access-token, for development. Tokens are cached until shortly before their real expiry and fetched once under concurrent demand. Disable with config :gcp_gcs, :gcloud_fallback, false; pin the binary with config :gcp_gcs, :gcloud_path, "/path/to/gcloud". A run that takes longer than config :gcp_gcs, :gcloud_timeout (default 30 000 ms) is killed.

A 401 response to an authenticated request is retried once with a fresh token: the rejected gcloud token is replaced (once, however many requests were rejected); Goth tokens are left to Goth.

Using Goth:

credentials = "GOOGLE_APPLICATION_CREDENTIALS_JSON" |> System.fetch_env!() |> JSON.decode!()

children = [
  {Goth,
   name: MyApp.Goth,
   source:
     {:service_account, credentials,
      scopes: ["https://www.googleapis.com/auth/devstorage.read_write"]}}
]

Call GcpGcs.clear_auth_cache/0 to force a token refresh after rotating credentials. It returns at once, and requests made after it fetch a new token, even while an earlier gcloud run is still in progress.

Telemetry

:telemetry.attach(
  "gcs-logger",
  [:gcp_gcs, :request, :stop],
  fn _event, %{duration: duration}, %{operation: op, result: result}, _config ->
    ms = System.convert_time_unit(duration, :native, :millisecond)
    require Logger
    Logger.info("gcs #{op} -> #{inspect(result)} in #{ms}ms")
  end,
  nil
)

Each public function emits one [:gcp_gcs, :request, _] span covering all of its HTTP calls (retries, chunks, rewrite steps); :operation names the function (:put_object, :upload_file, :upload_stream, :download, :download_to_file, :download_stream, :list_objects, …). Metadata always has :bucket and :object (the destination for copy/rewrite/move); upload_file/4 and download_to_file/4 also carry local file paths (:source/:destination). Treat telemetry metadata as potentially sensitive before exporting it.

[:gcp_gcs, :auth, _] spans are emitted only when the gcloud fallback actually runs gcloud (a cache miss or refresh), with source: :gcloud. Goth token reads emit nothing here; use Goth's own telemetry for its refreshes.

Testing

Unit tests run with no external services:

mix test

Integration tests run against a local emulator:

docker-compose up -d
mix test --include integration
docker-compose down

Why the JSON API (and not gRPC)?

Cloud Storage is a blob store: the dominant work is transferring object bytes, which maps naturally onto HTTP. Downloads stream as a response body and uploads use the documented resumable protocol — both with constant memory. The JSON API is the canonical, fully-featured Storage surface, and building on Finch keeps the dependency footprint small.

License

MIT — see LICENSE.

Links