Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381//! what every namespace shares: host resolution, auth, status-level retries,//! error mapping. one `Client` per process, borrowing a `zttp.Client` so//! connections are pooled across namespaces and threads and every phase of//! a request is bounded.
const Client = @This();
const std = @import("std");const json = std.json;const Allocator = std.mem.Allocator;const Io = std.Io;const zttp = @import("zttp");const Namespace = @import("Namespace.zig");
const log = std.log.scoped(.tpuf);
pub const version = "0.1.4";
transport: *zttp.Client,api_key: []const u8,host: Host,retry: Retry,
pub const Host = union(enum) { /// documented form: `https://{region}.turbopuffer.com` region: []const u8, /// any base url, no trailing slash base_url: []const u8,
/// the region-less host the production adopters were already using. pub const default: Host = .{ .base_url = "https://api.turbopuffer.com" };};
/// 429 and 5xx responses are retried with exponential backoff and equal/// jitter (half the step fixed, half random, drawn from the transport's/// `Io`), or after `Retry-After` when the api sends one. transport failures/// are not retried here: zttp already replays the ones that provably never/// took effect, and names the rest so the caller can decide.////// upsert, patch and delete are idempotent, so a retried write is safe./// a write carrying a `*_condition` is sent once: `Namespace.write` turns/// retries off for it at compile time.pub const Retry = struct { max_retries: u8 = 2, base_delay_ms: u32 = 250, max_delay_ms: u32 = 2_000, /// a `Retry-After` beyond this is clamped rather than waited out. max_retry_after_ms: u32 = 30_000,
pub const none: Retry = .{ .max_retries = 0 };
/// the capped exponential step for `attempt`, before jitter. pub fn stepMs(self: Retry, attempt: u8) u32 { const shift: u5 = @intCast(@min(attempt, 20)); const scaled: u64 = @as(u64, self.base_delay_ms) << shift; return @intCast(@min(scaled, self.max_delay_ms)); }
pub fn delayMs(self: Retry, random: std.Random, attempt: u8, retry_after_s: ?u64) u32 { if (retry_after_s) |s| { const ms = std.math.mul(u64, s, 1000) catch return self.max_retry_after_ms; return @intCast(@min(ms, self.max_retry_after_ms)); } const step = self.stepMs(attempt); const half = step / 2; return half + random.uintAtMost(u32, step - half); }};
pub const Options = struct { api_key: []const u8, host: Host = .default, retry: Retry = .{},};
/// errors this client adds over zttp's. a non-2xx status maps to one of/// these; `Diagnostics` carries the api's message.pub const Error = error{ BadRequest, Unauthorized, Forbidden, NotFound, Conflict, RateLimited, ServerError, UnexpectedStatus, UrlTooLong, ApiKeyTooLong,};
/// optional sink for the api's error body. pass one when the message matters/// (schema mismatches, conditional-write failures); otherwise the message is/// logged under the `tpuf` scope and the status is mapped to an error.pub const Diagnostics = struct { status: u16 = 0, buf: [512]u8 = undefined, len: usize = 0,
pub fn message(self: *const Diagnostics) []const u8 { return self.buf[0..self.len]; }
pub fn set(self: *Diagnostics, status: u16, body: []const u8) void { self.status = status; self.len = (errorMessage(&self.buf, body) orelse copyTruncated(&self.buf, body)).len; }};
pub fn init(transport: *zttp.Client, options: Options) Client { return .{ .transport = transport, .api_key = options.api_key, .host = options.host, .retry = options.retry, };}
pub fn namespace(self: *const Client, name: []const u8) error{InvalidNamespace}!Namespace { return Namespace.init(self, name);}
pub const ListOptions = struct { prefix: ?[]const u8 = null, page_size: ?u32 = null, cursor: ?[]const u8 = null, diagnostics: ?*Diagnostics = null,};
pub const NamespaceList = struct { namespaces: []const Entry = &.{}, next_cursor: ?[]const u8 = null,
pub const Entry = struct { id: []const u8 };};
/// one page of namespaces. pass `next_cursor` back as `cursor` for the next.pub fn listNamespaces(self: *const Client, alloc: Allocator, options: ListOptions) !json.Parsed(NamespaceList) { var path_buf: [1024]u8 = undefined; var w: std.Io.Writer = .fixed(&path_buf); w.writeAll("/v1/namespaces") catch return error.UrlTooLong; var sep: u8 = '?'; if (options.prefix) |prefix| { w.print("{c}prefix=", .{sep}) catch return error.UrlTooLong; percentEncode(&w, prefix) catch return error.UrlTooLong; sep = '&'; } if (options.page_size) |n| { w.print("{c}page_size={d}", .{ sep, n }) catch return error.UrlTooLong; sep = '&'; } if (options.cursor) |cursor| { w.print("{c}cursor=", .{sep}) catch return error.UrlTooLong; percentEncode(&w, cursor) catch return error.UrlTooLong; }
const body = try self.request(.GET, w.buffered(), null, .{ .diagnostics = options.diagnostics }); defer self.free(body); return json.parseFromSlice(NamespaceList, alloc, body, .{ .ignore_unknown_fields = true }) catch return error.UnexpectedResponse;}
pub const RequestOptions = struct { diagnostics: ?*Diagnostics = null, /// off for requests whose replay could apply twice: conditional writes. retry: bool = true,};
/// send one request and return the response body, owned by the transport's/// allocator: release it with `free`. `path` is absolute/// (`/v2/namespaces/foo/query`). retries per `self.retry`.////// `accept_encoding: identity` follows the api docs: "for most workloads,/// disabling compression offers the best performance. The official client/// libraries disable request and response compression by default."/// (https://turbopuffer.com/docs/api-overview#compression)pub fn request( self: *const Client, method: std.http.Method, path: []const u8, payload: ?[]const u8, options: RequestOptions,) ![]u8 { var url_buf: [1536]u8 = undefined; const url = try buildUrl(&url_buf, self.host, path);
var auth_buf: [256]u8 = undefined; const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{self.api_key}) catch return error.ApiKeyTooLong;
const max_retries: u8 = if (options.retry) self.retry.max_retries else 0; const random_source: std.Random.IoSource = .{ .io = self.transport.io };
var attempt: u8 = 0; while (true) : (attempt += 1) { var res = try self.transport.fetch(.{ .url = url, .method = method, .payload = payload, .authorization = auth, .accept_encoding = "identity", .content_type = if (payload != null) "application/json" else null, });
const status: u16 = @intFromEnum(res.status); if (status >= 200 and status < 300) { const body = res.body; res.body = &.{}; res.deinit(self.transport.allocator); return body; } defer res.deinit(self.transport.allocator);
if (isRetryable(status) and attempt < max_retries) { const now = Io.Clock.real.now(self.transport.io).toSeconds(); const ms = self.retry.delayMs(random_source.interface(), attempt, res.rate_limit.delaySeconds(now)); log.warn("{t} {s} -> {d}, retrying in {d}ms", .{ method, path, status, ms }); try self.transport.io.sleep(Io.Duration.fromMilliseconds(ms), .awake); continue; }
if (options.diagnostics) |d| d.set(status, res.body); var msg_buf: [200]u8 = undefined; log.err("{t} {s} -> {d}: {s}", .{ method, path, status, errorMessage(&msg_buf, res.body) orelse copyTruncated(&msg_buf, res.body) }); return statusError(status); }}
pub fn free(self: *const Client, body: []u8) void { self.transport.allocator.free(body);}
pub fn buildUrl(buf: []u8, host: Host, path: []const u8) error{UrlTooLong}![]const u8 { return switch (host) { .region => |region| std.fmt.bufPrint(buf, "https://{s}.turbopuffer.com{s}", .{ region, path }), .base_url => |base| std.fmt.bufPrint(buf, "{s}{s}", .{ base, path }), } catch error.UrlTooLong;}
pub fn isRetryable(status: u16) bool { return switch (status) { 429, 500, 502, 503, 504 => true, else => false, };}
pub fn statusError(status: u16) Error { return switch (status) { 400 => error.BadRequest, 401 => error.Unauthorized, 403 => error.Forbidden, 404 => error.NotFound, 409 => error.Conflict, 429 => error.RateLimited, 500...599 => error.ServerError, else => error.UnexpectedStatus, };}
const Envelope = struct { @"error": ?[]const u8 = null };
/// the api's error envelope is `{"status":"error","error":"..."}`. the/// message is copied into `out`, truncated to fit.pub fn errorMessage(out: []u8, body: []const u8) ?[]const u8 { var scratch: [2048]u8 = undefined; var fba: std.heap.FixedBufferAllocator = .init(&scratch); const envelope = json.parseFromSliceLeaky(Envelope, fba.allocator(), body, .{ .ignore_unknown_fields = true }) catch return null; return copyTruncated(out, envelope.@"error" orelse return null);}
fn copyTruncated(out: []u8, text: []const u8) []const u8 { const n = @min(out.len, text.len); @memcpy(out[0..n], text[0..n]); return out[0..n];}
/// query-string encoding: unreserved bytes pass through, everything else is %XX.pub fn percentEncode(w: *std.Io.Writer, s: []const u8) !void { for (s) |c| { switch (c) { 'A'...'Z', 'a'...'z', '0'...'9', '-', '_', '.', '~' => try w.writeByte(c), else => try w.print("%{X:0>2}", .{c}), } }}
test "buildUrl for region and base_url hosts" { var buf: [256]u8 = undefined; try std.testing.expectEqualStrings( "https://gcp-us-central1.turbopuffer.com/v2/namespaces/ns/query", try buildUrl(&buf, .{ .region = "gcp-us-central1" }, "/v2/namespaces/ns/query"), ); try std.testing.expectEqualStrings( "https://api.turbopuffer.com/v1/namespaces", try buildUrl(&buf, .default, "/v1/namespaces"), ); var tiny: [8]u8 = undefined; try std.testing.expectError(error.UrlTooLong, buildUrl(&tiny, .default, "/v1/namespaces"));}
test "status mapping and retryability" { try std.testing.expectEqual(error.BadRequest, statusError(400)); try std.testing.expectEqual(error.Conflict, statusError(409)); try std.testing.expectEqual(error.RateLimited, statusError(429)); try std.testing.expectEqual(error.ServerError, statusError(503)); try std.testing.expectEqual(error.UnexpectedStatus, statusError(302)); try std.testing.expect(isRetryable(429)); try std.testing.expect(isRetryable(502)); try std.testing.expect(!isRetryable(400)); try std.testing.expect(!isRetryable(409));}
test "errorMessage reads the api envelope" { var out: [64]u8 = undefined; try std.testing.expectEqualStrings( "attribute \"status\" not found in schema", errorMessage(&out, "{\"status\":\"error\",\"error\":\"attribute \\\"status\\\" not found in schema\"}").?, ); try std.testing.expectEqual(@as(?[]const u8, null), errorMessage(&out, "<html>bad gateway</html>")); try std.testing.expectEqual(@as(?[]const u8, null), errorMessage(&out, "{\"status\":\"ok\"}")); var small: [5]u8 = undefined; try std.testing.expectEqualStrings("attri", errorMessage(&small, "{\"error\":\"attribute\"}").?);}
test "Diagnostics captures status and message" { var d: Diagnostics = .{}; d.set(400, "{\"status\":\"error\",\"error\":\"duplicate id\"}"); try std.testing.expectEqual(@as(u16, 400), d.status); try std.testing.expectEqualStrings("duplicate id", d.message()); d.set(502, "upstream exploded"); try std.testing.expectEqualStrings("upstream exploded", d.message());}
test "Retry step doubles and caps" { const r: Retry = .{ .base_delay_ms = 100, .max_delay_ms = 350 }; try std.testing.expectEqual(@as(u32, 100), r.stepMs(0)); try std.testing.expectEqual(@as(u32, 200), r.stepMs(1)); try std.testing.expectEqual(@as(u32, 350), r.stepMs(2)); try std.testing.expectEqual(@as(u32, 350), r.stepMs(40));}
test "Retry delay jitters within [step/2, step] and defers to Retry-After" { const r: Retry = .{ .base_delay_ms = 100, .max_delay_ms = 350, .max_retry_after_ms = 5_000 }; var prng: std.Random.DefaultPrng = .init(42); const random = prng.random(); var seen_low = false; var seen_high = false; for (0..200) |_| { const ms = r.delayMs(random, 1, null); try std.testing.expect(ms >= 100 and ms <= 200); if (ms < 130) seen_low = true; if (ms > 170) seen_high = true; } try std.testing.expect(seen_low and seen_high); try std.testing.expectEqual(@as(u32, 3_000), r.delayMs(random, 0, 3)); try std.testing.expectEqual(@as(u32, 5_000), r.delayMs(random, 0, 3600)); try std.testing.expectEqual(@as(u32, 5_000), r.delayMs(random, 0, std.math.maxInt(u64)));}
test "NamespaceList parses the api's shape" { const parsed = try json.parseFromSlice(NamespaceList, std.testing.allocator, \\{"namespaces":[{"id":"bufos"},{"id":"bufos-v2"}],"next_cursor":null} , .{ .ignore_unknown_fields = true }); defer parsed.deinit(); try std.testing.expectEqual(@as(usize, 2), parsed.value.namespaces.len); try std.testing.expectEqualStrings("bufos-v2", parsed.value.namespaces[1].id); try std.testing.expectEqual(@as(?[]const u8, null), parsed.value.next_cursor);
const paged = try json.parseFromSlice(NamespaceList, std.testing.allocator, \\{"namespaces":[{"id":"a"}],"next_cursor":"eyJ"} , .{ .ignore_unknown_fields = true }); defer paged.deinit(); try std.testing.expectEqualStrings("eyJ", paged.value.next_cursor.?);}
test "percentEncode" { var buf: [64]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); try percentEncode(&w, "user-a b/c~"); try std.testing.expectEqualStrings("user-a%20b%2Fc~", w.buffered());}