`objectstore.http`

objectstore.http — an HTTP/HTTPS client over libcurl.

Why libcurl. Nothing in the Mojo package ecosystem gives this repo a TLS capable HTTP client: floki, flare, lightbug-http and fire-http all fail to resolve from conda ("No candidates were found") on both the stable and the nightly toolchain, and an object store without HTTPS is not an object store. conda-forge's libcurl ships TLS and a CA bundle for every platform we target, so this tin follows the magmalake FFI recipe (zstd.mojo, lz4.mojo): a tiny C shim built as a pixi-build-cmake package into $CONDA_PREFIX/lib, found by _dlopen.open_library (with or without CONDA_PREFIX) and dlopened once per process through an OwnedDLHandle.

The shim exists because curl_easy_setopt is a C variadic, which a dlopened symbol cannot express; shim/curl_wrapper.c collapses a whole request into one fixed-arity call and hands back an opaque result object read through accessors.

Every FFI-calling worker takes the handle as a borrowed imm parameter: Mojo destroys a value at its last syntactic use, so a function that both owned the handle and called get_function on it would dlclose the library before the returned function pointer is invoked.

Connection reuse. The curl easy handle lives in the shim and persists across requests, so a second request to the same host reuses its TCP and TLS session instead of shaking hands again — the difference between ~2.9 ms and ~0.2 ms per range read on loopback. Requests go to a process-wide pool by default; new_connection_pool hands out a private one for a caller that wants its connections kept apart.

Threads. Requests can be made from several threads at once (shim ABI 5): SHARED_POOL and a batch's multi handle are per thread in the shim, with their own connections, and what threads share — DNS and TLS sessions — sits behind locks. A private pool from new_connection_pool belongs to one thread at a time.

Batches. request_many sends several requests at once and returns when all have answered: one thread, curl's multi interface underneath, its own persistent connections. A ranged read of a Parquet file is mostly waiting for first bytes, so overlapping the waits is most of the speed there is to have.

Retries. Transport failures, 429 and most 5xx are retried with exponential backoff and jitter, but only for requests that can safely be repeated — see RetryPolicy. A response that is returned is never retried behind the caller's back on a method the server might have already acted on.

Structs

`Header`

struct Header

One Name: Value pair. Names are compared case-insensitively.

Implements Copyable, Movable, Writable

Fields: name: String; value: String

`write_to`

def write_to(self, mut writer: T)

`RetryPolicy`

struct RetryPolicy

When to try again, and how long to wait first.

Defaults: three retries (four attempts), 100 ms doubling to a 20 s ceiling, and idempotent requests only. POST is retried only when the caller has said the server can deduplicate it by supplying an Idempotency-Key header — which is exactly what an Iceberg REST commit does, and why the header is the trigger rather than a flag on the client.

Implements Copyable, Movable

Fields: max_retries: Int: Attempts after the first. 0 disables retrying entirely.; base_delay_ms: Int; max_delay_ms: Int; retry_non_idempotent: Bool: Escape hatch for a caller that knows its POST is safe to repeat and cannot add a header. Off, because guessing wrong duplicates a write.

`init`

def __init__(out self)

`none`

def none() -> Self

One request, one answer — the 0.1.x behaviour.

`Response`

struct Response

A completed HTTP response: status line, headers, and the whole body.

Implements Copyable, Movable

Fields: status: Int; headers: List[Header]; body: List[UInt8]

`init`

def __init__(out self, status: Int, var headers: List[Header], var body: List[UInt8])

`ok`

def ok(self) -> Bool

`header`

def header(self, name: String) -> String

The first value for name (case-insensitive), or "".

`has_header`

def has_header(self, name: String) -> Bool

`text`

def text(self) -> String

`body_excerpt`

def body_excerpt(self, limit: Int = Int(256)) -> String

The first limit bytes of the body, for error messages.

S3 and Iceberg REST both answer errors with a short XML or JSON document; truncating keeps a raised Error readable when the body is a 20 MB object instead.

`raise_for_status`

def raise_for_status(self, what: String)

`BatchRequest`

struct BatchRequest

One request of a batch: a GET or HEAD, optionally ranged.

expect_len, when known, sizes the response buffer up front.

Implements Copyable, Movable

Fields: method: String; url: String; headers: List[Header]; range_start: Int; range_end: Int; expect_len: Int

`get`

def get(var url: String, var headers: List[Header] = List(), range_start: Int = Int(-1), range_end: Int = Int(-1)) -> Self

`HttpClient`

struct HttpClient

An HTTP client over a persistent, reused connection.

The client itself holds no socket: pool names a curl handle living in the shim, and SHARED_POOL — the default — is the process-wide one. Two HttpClient values therefore share connections, which is what makes an Iceberg scan that opens a new S3Client per file cheap. With the default pool, threads may each make requests at once.

Redirects are off by default: a presigned S3 URL carries its signature in the query string and must never be silently re-issued elsewhere.

Implements Copyable, Movable

Fields: timeout_ms: Int; follow_redirects: Bool; verify_tls: Bool; pool: Int: SHARED_POOL, or a handle from new_connection_pool() that the caller is responsible for freeing.; retry: RetryPolicy

`init`

def __init__(out self)

`request`

def request(self, method: String, url: String, headers: List[Header] = List(), body: Span[UInt8] = Span(), range_start: Int = Int(-1), range_end: Int = Int(-1)) -> Response

Performs the request, retrying per self.retry.

Raises only on transport failure — a 404 or a 500 comes back as a Response so the caller can decide. A retryable status that survives the last attempt is returned too: the caller sees the server's own answer rather than an invented error.

`request_many`

def request_many(self, reqs: List[BatchRequest], max_parallel: Int = Int(16)) -> List[Response]

Every request in reqs at once, answers in the same order.

Like request, a status comes back as a Response and only a transport failure raises. A request that failed in transport or got a retryable answer is tried again on its own, under self.retry. Batches are GETs and HEADs, which are always safe to repeat. A shim older than ABI 4 has no batches: then they run one by one.

`get`

def get(self, url: String, headers: List[Header] = List(), range_start: Int = Int(-1), range_end: Int = Int(-1)) -> Response

`head`

def head(self, url: String, headers: List[Header] = List()) -> Response

`put`

def put(self, url: String, body: Span[UInt8], headers: List[Header] = List()) -> Response

`post`

def post(self, url: String, body: Span[UInt8], headers: List[Header] = List()) -> Response

`delete`

def delete(self, url: String, headers: List[Header] = List()) -> Response

Functions

`new_connection_pool`

def new_connection_pool() -> Int

A private pool: one persistent curl handle, its own connections.

Returns an opaque handle to pass as HttpClient.pool. The caller owns it and must call free_connection_pool; there is no destructor here because HttpClient is Copyable, and a copyable owner of a C resource is a double free waiting to happen. Most callers want SHARED_POOL.

`free_connection_pool`

def free_connection_pool(pool: Int)

Closes a pool from new_connection_pool and its live connections.

`pool_stats`

def pool_stats(pool: Int = Int(0)) -> Tuple[Int, Int]

(requests, connections) for a pool since the last reset_pool_stats.

connections is what curl actually opened, so reuse is the observation that it stays at 1 while requests climbs — which is how the tests prove it rather than inferring it from a stopwatch.

`reset_pool_stats`

def reset_pool_stats(pool: Int = Int(0))

`shim_abi`

def shim_abi() -> Int

1 = a fresh curl handle per request (0.1.x); 2 = pooled connections; 3 = instance_credentials_blob; 4 = batches (request_many); 5 = safe from several threads.

`instance_credentials_blob`

def instance_credentials_blob() -> String

The EC2 instance role's credentials, as the shim's IMDSv2 lookup hands them back: access-key-id secret token expiry-epoch, or "" when there are none.

The shim caches the answer for the process, "none" included, so this is cheap to call per request. A shim older than ABI 3 has no lookup and answers "" as well.

`header_blob`

def header_blob(headers: List[Header]) -> String

Headers as the shim wants them: Name: Value lines joined by LF.

An HTTP header value cannot contain a bare LF, so this encoding is lossless and spares the FFI boundary a char*[].

`is_idempotent`

def is_idempotent(method: String, headers: List[Header]) -> Bool

RFC 9110 idempotency, plus the Idempotency-Key convention.

GET, HEAD, PUT, DELETE, OPTIONS and TRACE are idempotent by definition: repeating one cannot commit a second effect. POST and PATCH are not, so they are only repeatable when the request carries a key the server can deduplicate on.

`retryable_status`

def retryable_status(status: Int) -> Bool

429 and 5xx, except the two that mean "never, no matter how often".

501 Not Implemented and 505 HTTP Version Not Supported are permanent facts about the server; retrying them is pure latency.

`s3_retryable_error`

def s3_retryable_error(body: Span[UInt8]) -> Bool

S3's two "come back later" errors, which arrive with unhelpful statuses.

SlowDown is throttling and RequestTimeout means the server gave up waiting on the body; AWS returns the latter as a 400, so status alone would call it permanent. Only the first bytes are inspected, and only for a response that already failed, so this never touches an object body.

`backoff_delay_ms`

def backoff_delay_ms(policy: RetryPolicy, attempt: Int) -> Int

Exponential backoff with jitter: half the window fixed, half random.

attempt counts from 1. Full jitter would sometimes retry instantly and hammer a server that is already asking for room; half is the usual compromise. The jitter source is the monotonic clock's low bits — there is no RNG in reach here, and a retry schedule does not need a good one.

`parse_headers`

def parse_headers(raw: String) -> List[Header]

Parses curl's raw header text into pairs.

curl emits one header block per response it sees, so a redirect chain or a 100 Continue produces several. Only the last block describes the response the caller gets, so a new status line resets the accumulator.

`batch_stats`

def batch_stats() -> Tuple[Int, Int]

(requests, connections) for batches since reset_batch_stats: the connections stay few while requests climb, because batches reuse them.

`reset_batch_stats`

def reset_batch_stats()

`curl_version`

def curl_version() -> String

Libcurl's own version banner, e.g. libcurl/8.9.1 OpenSSL/3.3.1 ….

Aliases

comptime DEFAULT_MAX_PARALLEL = 16

Requests a batch keeps in flight at once, on as many connections. The Rust Iceberg reader keeps 10 range reads in flight per file; S3 serves each connection at roughly 100 MB/s, so 16 covers a node's share of bandwidth without opening sockets by the hundred.

comptime DEFAULT_TIMEOUT_MS = 30000

Total request deadline. Range reads of a Parquet footer are small; a 20 MB object over a slow link is not, so this is generous rather than tight.

comptime SHARED_POOL = 0

The process-wide connection pool, which is what every HttpClient uses unless it is given one of its own.

comptime S3_RETRYABLE_CODES = 2