Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616//! resync — whole-repo replacement for a previously failed/pending repo//!//! upstream's HandleRepoResync (docs/upstream-bootstrap-spec.md): download//! the repo via the relay and emit one KindSync DID tombstone followed by//! KindCreateResync rows for the repo's current records. compaction later//! treats the sync row as a DID tombstone (everything below it folds away)//! and the resync rows as the fresh materialization above it.//!//! atomicity mirrors upstream's pre-validation: the CAR is parsed, indexed,//! and completeness-checked BEFORE anything is appended. Rows then stream//! through bounded 1,024-row batches, matching upstream. This bounds memory//! without holding the shared writer across an arbitrarily large repository;//! live ingestion may proceed between batches.
const std = @import("std");const zat = @import("zat");const archive_mod = @import("../../storage/archive.zig");const observability = @import("../../runtime/observability.zig");const fetched_car = @import("fetched_car.zig");const prepared_fetch = @import("prepared_fetch.zig");const repo_store = @import("repo_store.zig");const repos = @import("repos.zig");const segment = @import("../../storage/segment.zig");const segment_writer = @import("../../storage/segment_writer.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);
pub const Stats = struct { rows: u64 = 0,};
pub const AttemptDetails = struct { final_host: fetched_car.FinalHost = .{}, retry_after_s: ?u64 = null, error_class: repo_store.ErrorClass = .unknown, /// True when the relay's 302 served (or attempted) the download instead /// of the stamped PDS: either no routable stamp existed, or the stamped /// PDS authoritatively lacked the repo (stale, pre-migration stamp). via_fallback: bool = false, /// True once the network/CAR input has been accepted and the failure is /// attributable to Stream's archive or metadata path. Callers must abort /// lifecycle work instead of recording this as a remote repo failure. local_failure: bool = false,};
/// download `did` via the relay and append its replacement stream/// (sync tombstone + create_resync rows) to `archive`, marking the repo/// complete in `store`. malformed repositories fail before the transaction;/// local writer failures are fatal; successful rows become durable before the/// repository completion watermark.pub fn resyncRepo( allocator: Allocator, io: Io, scratch_dir: []const u8, relay_http: []const u8, pds: []const u8, did: []const u8, archive: *archive_mod.Archive, store: *repo_store.Store, drops: *repos.DropCounts, details: *AttemptDetails, trace_context: observability.Context,) !Stats { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const alloc = arena.allocator();
const previous = store.get(allocator, did) catch |err| { details.local_failure = true; return err; }; defer if (previous) |state| store.freeState(allocator, state); const active = if (previous) |state| state.active else true; // PDS-direct first, mirroring upstream retryRunner.download: an // unroutable stamp falls back to the relay's 302 immediately, and a // direct RepoNotFound falls back too — the stamped PDS authoritatively // lacking the repo means a stale pre-migration stamp, and without the // fallback that not_found would mark an undownloaded migrated repo // complete below. Other direct failures (5xx, timeout) stay ordinary // retry failures with backoff, exactly like upstream. var base_buf: [std.Io.net.HostName.max_len + 16]u8 = undefined; const direct = directBase(pds, &base_buf); details.via_fallback = direct == null; var outcome = prepared_fetch.fetch(allocator, alloc, io, scratch_dir, direct orelse relay_http, did, &details.final_host) catch |err| { details.error_class = repo_store.classifyAttemptError(err); return err; }; if (!details.via_fallback and outcome == .not_found) { details.via_fallback = true; details.final_host = .{}; outcome = prepared_fetch.fetch(allocator, alloc, io, scratch_dir, relay_http, did, &details.final_host) catch |err| { details.error_class = repo_store.classifyAttemptError(err); return err; }; } var prepared_fetch_result = switch (outcome) { .ok => |value| value, .rate_limited => |retry_after| { details.retry_after_s = retry_after; details.error_class = .http_429; return error.RateLimited; }, .not_found => { store.put(did, terminalState(previous, .complete, active, details.final_host.slice(), Io.Timestamp.now(io, .real).toMicroseconds())) catch |err| { details.local_failure = true; return err; }; return .{}; }, .unavailable => { store.put(did, terminalState(previous, .unavailable, active, details.final_host.slice(), Io.Timestamp.now(io, .real).toMicroseconds())) catch |err| { details.local_failure = true; return err; }; return .{}; }, .transient => { details.error_class = .http_5xx; return error.GetRepoFailed; }, .permanent => { details.error_class = .unknown; return error.GetRepoFailed; }, }; defer prepared_fetch_result.deinit(io); // The direct path routes to untrusted PDSes and bypasses identity // resolution: a CAR whose commit identifies a different DID must not be // resynced under `did` (upstream retryRunner.download's mismatch guard). if (!std.mem.eql(u8, prepared_fetch_result.repo.embedded_did, did)) { details.error_class = .unknown; return error.GetRepoDidMismatch; } details.local_failure = true;
const attributes = [_]observability.Attribute{ .{ .key = "did", .value = .{ .string = did } }, }; var trace = observability.Span.start(allocator, "ingest/backfill", "handleRepo", trace_context, &attributes); defer trace.deinit(); defer trace.succeed(); return trace.recordError(appendReplacement( alloc, io, pds, did, &prepared_fetch_result.repo, archive, store, drops, details, previous, active, trace.context(), ));}
fn appendReplacement( alloc: Allocator, io: Io, pds: []const u8, did: []const u8, prepared_repo: *repos.PreparedRepo, archive: *archive_mod.Archive, store: *repo_store.Store, drops: *repos.DropCounts, details: *AttemptDetails, previous: ?repo_store.RepoState, active: bool, trace_context: observability.Context,) !Stats { const now_us: i64 = Io.Timestamp.now(io, .real).toMicroseconds();
const head_rev = prepared_repo.rev;
const sync_ev = try syncTombstoneEvent(alloc, did, head_rev, now_us);
const batch_size = 1024; const Sink = struct { archive: *archive_mod.Archive, allocator: Allocator, now_us: i64, batch: std.ArrayList(segment.Event) = .empty, row_count: u64 = 0, trace_context: observability.Context,
fn flush(self: *@This()) anyerror!void { if (self.batch.items.len == 0) return; _ = try self.archive.appendBatchTraced(self.batch.items, self.now_us, self.trace_context); self.batch.clearRetainingCapacity(); }
fn emit(self: *@This(), event: segment.Event) anyerror!void { try self.batch.append(self.allocator, event); self.row_count += 1; if (self.batch.items.len == batch_size) try self.flush(); } }; var sink: Sink = .{ .archive = archive, .allocator = alloc, .now_us = now_us, .trace_context = trace_context, }; defer sink.batch.deinit(alloc); try sink.emit(sync_ev); _ = try repos.emitPrepared(alloc, prepared_repo, did, now_us, .create_resync, drops, &sink, Sink.emit); try sink.flush(); try archive.flushBlockTraced(trace_context);
// durability ordering: every replacement batch is durable before the // completion row that describes it. const completed_us = Io.Timestamp.now(io, .real).toMicroseconds(); try store.put(did, .{ .status = .complete, .active = active, .rev = head_rev, .latest_rev = head_rev, .host = details.final_host.slice(), .handle = if (previous) |state| state.handle else "", // Fallback success means the stamp was stale or absent: repair it to // the host that actually served the CAR so future passes go direct // (upstream's restamp via updateRepoHostActive, guarded by // completionHost != cand.PDS). https because upstream's stamps are // bare authorities later dialed by atmos's hardened https-only // client builder (run.go NewHostClientBuilder). .pds = if (shouldRestamp(details.via_fallback, details.final_host.slice(), pds)) try std.fmt.allocPrint(alloc, "https://{s}", .{details.final_host.slice()}) else if (previous) |state| state.pds else "", .attempts = 0, .last_attempt_us = completed_us, .started_us = if (previous) |state| state.started_us else 0, .completed_us = completed_us, .retry_count = 0, .next_attempt_us = 0, .updated_us = completed_us, .record_count = if (previous) |state| state.record_count else 0, .total_bytes = if (previous) |state| state.total_bytes else 0, }); try store.flush(); return .{ .rows = sink.row_count + 1 };}
/// Base URL for the direct getRepo attempt, or null when the stamp is/// unroutable (upstream's clientForHost build-failure path). Upstream stamps/// bare authorities (hostBucketFromAuthority) and dials scheme://authority;/// our stamp is the DID document's #atproto_pds URL, so this reduces it to/// the same shape — stamped scheme + normalized authority, path discarded.pub fn directBase(pds: []const u8, buf: []u8) ?[]const u8 { const uri = std.Uri.parse(pds) catch return null; if (!std.ascii.eqlIgnoreCase(uri.scheme, "https") and !std.ascii.eqlIgnoreCase(uri.scheme, "http")) return null; var authority: fetched_car.FinalHost = .{}; fetched_car.captureFinalHost(uri, &authority) catch return null; if (authority.slice().len == 0) return null; return std.fmt.bufPrint(buf, "{s}://{s}", .{ uri.scheme, authority.slice() }) catch null;}
/// Upstream's restamp guard, exactly: viaFallback && completionHost != ""/// && completionHost != cand.PDS — with cand.PDS compared as a normalized/// authority, since that is what upstream stamps.pub fn shouldRestamp(via_fallback: bool, final_host: []const u8, pds: []const u8) bool { if (!via_fallback or final_host.len == 0) return false; const uri = std.Uri.parse(pds) catch return true; var stamped: fetched_car.FinalHost = .{}; fetched_car.captureFinalHost(uri, &stamped) catch return true; return !std.mem.eql(u8, stamped.slice(), final_host);}
fn terminalState(previous: ?repo_store.RepoState, status: repo_store.Status, active: bool, host: []const u8, terminal_us: i64) repo_store.RepoState { return .{ .status = status, .active = active, .rev = if (previous) |state| state.rev else "", .latest_rev = if (previous) |state| state.latest_rev else "", .host = if (host.len > 0) host else if (previous) |state| state.host else "", .handle = if (previous) |state| state.handle else "", .pds = if (previous) |state| state.pds else "", .attempts = 0, .last_attempt_us = terminal_us, .started_us = if (previous) |state| state.started_us else 0, .completed_us = if (status == .complete) terminal_us else if (previous) |state| state.completed_us else 0, .retry_count = 0, .next_attempt_us = 0, .updated_us = if (previous) |state| state.updated_us else 0, .record_count = if (previous) |state| state.record_count else 0, .total_bytes = if (previous) |state| state.total_bytes else 0, };}
/// synthetic #sync payload matching what the live path archives (a CBOR/// map the cold reader pulls did/rev/time from; seq and blocks default)pub fn syncTombstoneEvent(alloc: Allocator, did: []const u8, rev: []const u8, now_us: i64) !segment.Event { var time_buf: [40]u8 = undefined; const time_str = try alloc.dupe(u8, formatRfc3339(&time_buf, now_us)); const entries = try alloc.dupe(zat.cbor.Value.MapEntry, &.{ .{ .key = "did", .value = .{ .text = did } }, .{ .key = "rev", .value = .{ .text = rev } }, .{ .key = "time", .value = .{ .text = time_str } }, }); const payload = try zat.cbor.encodeAlloc(alloc, .{ .map = entries });
const ev: segment.Event = .{ .seq = 0, // archive assigns .witnessed_at = now_us, .indexed_at = 0, .kind = .sync, .did = did, .collection = "", .rkey = "", .rev = rev, .payload = payload, }; try segment_writer.validateFields(ev); return ev;}
/// Preserve the original typed #sync envelope when an upstream sync triggered/// the repair. The envelope is client-visible on v2 and part of the durable/// event log; replacing it with a synthetic did/rev/time map loses its CAR,/// upstream seq, and original timestamp.pub fn syncTombstoneFromPayload(alloc: Allocator, did: []const u8, payload: []const u8, now_us: i64) !segment.Event { const envelope = try zat.cbor.decodeAll(alloc, payload); const rev = envelope.getString("rev") orelse return error.InvalidSyncEnvelope; const ev: segment.Event = .{ .seq = 0, .witnessed_at = now_us, .indexed_at = 0, .kind = .sync, .did = did, .collection = "", .rkey = "", .rev = rev, .payload = payload, }; try segment_writer.validateFields(ev); return ev;}
/// microseconds since epoch → "YYYY-MM-DDTHH:MM:SS.ssssssZ"pub fn formatRfc3339(buf: []u8, micros: i64) []const u8 { const text = @import("zat").Datetime.fromMicros(micros).toString(); @memcpy(buf[0..text.len], &text); return buf[0..text.len];}
// === tests ===
const testing = std.testing;
test "terminal no-repo resync states match upstream OnFail bookkeeping" { const previous: repo_store.RepoState = .{ .status = .failed, .active = true, .rev = "initial", .latest_rev = "latest", .host = "old.example", .handle = "alice.test", .pds = "https://identity.example", .last_error = "old failure", .error_class = .timeout, .attempts = 4, .started_us = 3, .completed_us = 5, .retry_count = 6, .next_attempt_us = 7, .updated_us = 8, .record_count = 9, .total_bytes = 10, }; const missing = terminalState(previous, .complete, true, "new.example", 100); try testing.expectEqual(@as(u32, 0), missing.attempts); try testing.expectEqual(@as(u32, 0), missing.retry_count); try testing.expectEqual(@as(i64, 0), missing.next_attempt_us); try testing.expectEqual(@as(i64, 100), missing.completed_us); try testing.expectEqual(@as(i64, 8), missing.updated_us); try testing.expectEqualStrings("initial", missing.rev); try testing.expectEqualStrings("latest", missing.latest_rev); try testing.expectEqualStrings("", missing.last_error); try testing.expectEqualStrings("alice.test", missing.handle);
const unavailable = terminalState(previous, .unavailable, true, "", 101); try testing.expectEqual(@as(i64, 5), unavailable.completed_us); try testing.expectEqual(@as(i64, 101), unavailable.last_attempt_us); try testing.expectEqualStrings("old.example", unavailable.host); try testing.expectEqual(@as(u32, 0), unavailable.attempts); try testing.expectEqual(@as(i64, 9), unavailable.record_count); try testing.expectEqual(@as(i64, 10), unavailable.total_bytes);}
test "formatRfc3339" { var buf: [40]u8 = undefined; // 1783900800 = 2026-07-13T00:00:00Z try testing.expectEqualStrings( "2026-07-13T00:00:01.500000Z", formatRfc3339(&buf, (1783900800 + 1) * std.time.us_per_s + 500_000), ); try testing.expectEqualStrings("1970-01-01T00:00:00.000000Z", formatRfc3339(&buf, 0));}
test "directBase reduces the stamped URL to scheme://authority, like upstream's bare-authority stamps" { var buf: [std.Io.net.HostName.max_len + 16]u8 = undefined; try testing.expectEqualStrings("https://pds.example", directBase("https://pds.example", &buf).?); // normalized like FinalHost: lowercased, default port dropped, non-default kept try testing.expectEqualStrings("https://pds.example:8443", directBase("https://PDS.EXAMPLE:8443", &buf).?); try testing.expectEqualStrings("https://pds.example", directBase("https://pds.example:443", &buf).?); // path and trailing slash discarded — upstream dials the authority only try testing.expectEqualStrings("https://pds.example", directBase("https://pds.example/", &buf).?); try testing.expectEqualStrings("http://127.0.0.1:3000", directBase("http://127.0.0.1:3000", &buf).?); try testing.expect(directBase("", &buf) == null); try testing.expect(directBase("unknown", &buf) == null); try testing.expect(directBase("pds.example", &buf) == null); try testing.expect(directBase("wss://pds.example", &buf) == null); try testing.expect(directBase("https://", &buf) == null);}
test "shouldRestamp mirrors upstream: viaFallback && completionHost != \"\" && completionHost != cand.PDS" { // direct success never restamps try testing.expect(!shouldRestamp(false, "pds.example", "https://old.example")); // fallback without an attributable host never restamps try testing.expect(!shouldRestamp(true, "", "https://old.example")); // fallback to a different host repairs the stamp try testing.expect(shouldRestamp(true, "pds.example", "https://old.example")); // fallback landing on the very host already stamped is not a repair try testing.expect(!shouldRestamp(true, "pds.example", "https://pds.example")); try testing.expect(!shouldRestamp(true, "pds.example", "https://PDS.EXAMPLE:443/")); // no or unroutable stamp: any served host is a gained route try testing.expect(shouldRestamp(true, "pds.example", "")); try testing.expect(shouldRestamp(true, "pds.example", "unknown"));}
const meta_store = @import("../../storage/meta_store.zig");
/// full signed single-record repo CAR, same fixture shape as repos.zig testsfn buildRepoCar(alloc: Allocator, secret: u8) !struct { car: []const u8, did: []const u8 } { const record = "pds-direct fallback fixture"; const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); var tree = zat.mst.Mst.init(alloc); try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid); const data_cid = try tree.rootCid(); const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(secret)); const did = try keypair.did(alloc); const signed = try zat.signCommit(alloc, .{ .did = did, .rev = "3k2abcdefghij", .data = data_cid, }, &keypair); var blocks: std.ArrayList(zat.car.Block) = .empty; try blocks.append(alloc, .{ .cid_raw = signed.cid.raw, .data = signed.bytes }); try tree.collectBlocks(&blocks); try blocks.append(alloc, .{ .cid_raw = record_cid.raw, .data = record }); const car = try zat.car.writeAlloc(alloc, .{ .roots = &.{signed.cid}, .blocks = blocks.items, }); return .{ .car = car, .did = did };}
const FixtureResponse = union(enum) { not_found, car: []const u8,};
fn serveOnce(listener: *Io.net.Server, task_io: Io, response: *const FixtureResponse) !void { var stream = try listener.accept(task_io); defer stream.close(task_io); var read_buf: [4096]u8 = undefined; var write_buf: [64 * 1024]u8 = undefined; var reader = stream.reader(task_io, &read_buf); var writer = stream.writer(task_io, &write_buf); var http_server = std.http.Server.init(&reader.interface, &writer.interface); _ = try http_server.receiveHead(); switch (response.*) { .not_found => try writer.interface.writeAll( "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", ), .car => |bytes| { try writer.interface.print( "HTTP/1.1 200 OK\r\nContent-Type: application/vnd.ipld.car\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n", .{bytes.len}, ); try writer.interface.writeAll(bytes); }, } try writer.interface.flush();}
// Mirrors upstream retryRunner.download's stale-stamp path: the stamped PDS// authoritatively lacks the repo (migration), the relay fallback serves it,// and the completion restamps the routing so the next pass goes direct.// Without the fallback, the direct 404 would have recorded an undownloaded// migrated repo as terminal-complete.test "stale PDS stamp falls back to the relay and restamps on success" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const alloc = arena.allocator(); const fixture = try buildRepoCar(alloc, 21);
var stale_pds = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer stale_pds.deinit(io); var relay = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer relay.deinit(io); const stale_response: FixtureResponse = .not_found; const relay_response: FixtureResponse = .{ .car = fixture.car }; var stale_future = try io.concurrent(serveOnce, .{ &stale_pds, io, &stale_response }); defer _ = stale_future.cancel(io) catch {}; var relay_future = try io.concurrent(serveOnce, .{ &relay, io, &relay_response }); defer _ = relay_future.cancel(io) catch {};
var archive_tmp = testing.tmpDir(.{}); defer archive_tmp.cleanup(); var meta_tmp = testing.tmpDir(.{}); defer meta_tmp.cleanup(); var scratch_tmp = testing.tmpDir(.{}); defer scratch_tmp.cleanup(); var archive_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; var meta_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; var scratch_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const archive_path = archive_path_buf[0..try archive_tmp.dir.realPath(io, &archive_path_buf)]; const meta_path = meta_path_buf[0..try meta_tmp.dir.realPath(io, &meta_path_buf)]; const scratch_path = scratch_path_buf[0..try scratch_tmp.dir.realPath(io, &scratch_path_buf)];
var archive = try archive_mod.Archive.init(testing.allocator, io, archive_path); defer archive.deinit(); var meta = try meta_store.Store.open(testing.allocator, meta_path); defer meta.deinit(); var store = try repo_store.Store.init(testing.allocator, io, &meta); defer store.deinit();
var url_buf: [96]u8 = undefined; const stale_url = try std.fmt.allocPrint(alloc, "http://127.0.0.1:{d}", .{stale_pds.socket.address.getPort()}); const relay_url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{relay.socket.address.getPort()}); try store.put(fixture.did, .{ .status = .failed, .pds = stale_url, .host = "stale.example" });
var drops: repos.DropCounts = .{}; var details: AttemptDetails = .{}; const stats = try resyncRepo(testing.allocator, io, scratch_path, relay_url, stale_url, fixture.did, &archive, &store, &drops, &details, observability.root_context); try testing.expect(stats.rows > 1); // sync tombstone + resync rows, not a terminal no-op try testing.expect(details.via_fallback);
const state = (try store.get(testing.allocator, fixture.did)).?; defer store.freeState(testing.allocator, state); try testing.expectEqual(repo_store.Status.complete, state.status); var want_pds_buf: [96]u8 = undefined; const want_pds = try std.fmt.bufPrint(&want_pds_buf, "https://127.0.0.1:{d}", .{relay.socket.address.getPort()}); try testing.expectEqualStrings(want_pds, state.pds);}
// Upstream retryRunner.download's mismatch guard: the direct path routes to// untrusted PDSes, so a CAR whose commit identifies a different DID must be// rejected rather than resynced under the requested DID.test "direct getRepo CAR with a foreign embedded DID is rejected" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const alloc = arena.allocator(); const fixture = try buildRepoCar(alloc, 22);
var pds = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer pds.deinit(io); const pds_response: FixtureResponse = .{ .car = fixture.car }; var pds_future = try io.concurrent(serveOnce, .{ &pds, io, &pds_response }); defer _ = pds_future.cancel(io) catch {};
var archive_tmp = testing.tmpDir(.{}); defer archive_tmp.cleanup(); var meta_tmp = testing.tmpDir(.{}); defer meta_tmp.cleanup(); var scratch_tmp = testing.tmpDir(.{}); defer scratch_tmp.cleanup(); var archive_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; var meta_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; var scratch_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const archive_path = archive_path_buf[0..try archive_tmp.dir.realPath(io, &archive_path_buf)]; const meta_path = meta_path_buf[0..try meta_tmp.dir.realPath(io, &meta_path_buf)]; const scratch_path = scratch_path_buf[0..try scratch_tmp.dir.realPath(io, &scratch_path_buf)];
var archive = try archive_mod.Archive.init(testing.allocator, io, archive_path); defer archive.deinit(); var meta = try meta_store.Store.open(testing.allocator, meta_path); defer meta.deinit(); var store = try repo_store.Store.init(testing.allocator, io, &meta); defer store.deinit();
var url_buf: [96]u8 = undefined; const pds_url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{pds.socket.address.getPort()}); var drops: repos.DropCounts = .{}; var details: AttemptDetails = .{}; try testing.expectError( error.GetRepoDidMismatch, resyncRepo(testing.allocator, io, scratch_path, "http://relay.invalid", pds_url, "did:plc:someoneelse", &archive, &store, &drops, &details, observability.root_context), ); try testing.expect(!details.local_failure); try testing.expect(!details.via_fallback);}
test "syncTombstoneEvent round-trips through the cold reader's fields" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const ev = try syncTombstoneEvent(arena.allocator(), "did:plc:abc", "3l3qo2vutsw2b", 1_750_000_000_000_000); try testing.expectEqual(segment.Kind.sync, ev.kind); try testing.expectEqualStrings("3l3qo2vutsw2b", ev.rev); const env = try zat.cbor.decodeAll(arena.allocator(), ev.payload); try testing.expectEqualStrings("did:plc:abc", env.getString("did").?); try testing.expectEqualStrings("3l3qo2vutsw2b", env.getString("rev").?); try testing.expect(env.getString("time").?.len == 27);}