From a2652c444200794a9fc0ff9ad24c37610ea885ff Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 16 Aug 2026 01:15:46 +0000 Subject: [PATCH] retry: match upstream's restamp guard and bare-authority dialing exactly review against retry.go found three divergences in the PDS-direct port; all removed rather than rationalized. restamp is now guarded by upstream's completionHost != cand.PDS (as a normalized authority comparison); the direct base reduces the stamped DID-document URL to scheme://normalized-authority, matching upstream dialing bare authorities; the https restamp is kept and documented as parity with atmos's hardened https-only host client builder, in docs/semantic-parity.md alongside the one recorded data-model difference (URL stamps vs authority stamps). Co-Authored-By: Claude Fable 5 --- docs/semantic-parity.md | 11 +++++++++++ src/internal/ingest/backfill/fetched_car.zig | 2 +- src/internal/ingest/backfill/resync.zig | 84 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------------- 3 file(s) changed, 73 insertion(s)(+), 24 deletion(s)(-) diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -14,6 +14,17 @@ > PDS-direct fleet bootstrap and its four backfill flags > (environment_contract.py `upstream_unported`) — Stream's bootstrap > remains relay-getRepo until that design is ported on purpose. +> +> **Failed-repo retry routing (2026-08-15, task #11):** the retry pass now +> ports `retryRunner.download` (retry.go:425-438) exactly — stamped-PDS +> direct first; relay 302 fallback only on an unroutable stamp or a direct +> `isRepoNotFoundError`; restamp guarded by upstream's +> `completionHost != cand.PDS`; foreign embedded DID rejected. One recorded +> data-model difference, not a behavioral one: upstream stamps bare +> authorities and dials them via atmos's https-only hardened client builder, +> while Stream stamps the DID document's `#atproto_pds` URL — `directBase` +> reduces it to `scheme://normalized-authority` before dialing, and restamps +> as `https://authority`, matching what upstream's builder would dial. ## Audit basis diff --git a/src/internal/ingest/backfill/fetched_car.zig b/src/internal/ingest/backfill/fetched_car.zig --- a/src/internal/ingest/backfill/fetched_car.zig +++ b/src/internal/ingest/backfill/fetched_car.zig @@ -331,7 +331,7 @@ return try task.result.?; } -fn captureFinalHost(uri: std.Uri, out: *FinalHost) !void { +pub fn captureFinalHost(uri: std.Uri, out: *FinalHost) !void { var host_buf: [std.Io.net.HostName.max_len]u8 = undefined; const host = try uri.getHost(&host_buf); const default_port: ?u16 = if (std.ascii.eqlIgnoreCase(uri.scheme, "https")) 443 else if (std.ascii.eqlIgnoreCase(uri.scheme, "http")) 80 else null; diff --git a/src/internal/ingest/backfill/resync.zig b/src/internal/ingest/backfill/resync.zig --- a/src/internal/ingest/backfill/resync.zig +++ b/src/internal/ingest/backfill/resync.zig @@ -80,8 +80,10 @@ // 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. - details.via_fallback = !validPdsBase(pds); - var outcome = prepared_fetch.fetch(allocator, alloc, io, scratch_dir, if (details.via_fallback) relay_http else pds, did, &details.final_host) catch |err| { + 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; }; @@ -192,8 +194,11 @@ .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). - .pds = if (details.via_fallback and details.final_host.slice().len > 0) + // (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, @@ -210,16 +215,29 @@ return .{ .rows = sink.row_count + 1 }; } -/// A stamp is routable when it is an absolute http(s) base URL with a host, -/// the shape engine.zig persists from the DID document's #atproto_pds -/// endpoint. Anything else falls back to the relay, like upstream's -/// clientForHost build-failure path. -pub fn validPdsBase(pds: []const u8) bool { - const uri = std.Uri.parse(pds) catch return false; - if (!std.ascii.eqlIgnoreCase(uri.scheme, "https") and !std.ascii.eqlIgnoreCase(uri.scheme, "http")) return false; - var host_buf: [std.Io.net.HostName.max_len]u8 = undefined; - const host = uri.getHost(&host_buf) catch return false; - return host.bytes.len > 0; +/// 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 { @@ -351,15 +369,35 @@ try testing.expectEqualStrings("1970-01-01T00:00:00.000000Z", formatRfc3339(&buf, 0)); } -test "validPdsBase accepts the DID-document endpoint shape and nothing else" { - try testing.expect(validPdsBase("https://pds.example")); - try testing.expect(validPdsBase("https://PDS.EXAMPLE:8443")); - try testing.expect(validPdsBase("http://127.0.0.1:3000")); - try testing.expect(!validPdsBase("")); - try testing.expect(!validPdsBase("unknown")); - try testing.expect(!validPdsBase("pds.example")); - try testing.expect(!validPdsBase("wss://pds.example")); - try testing.expect(!validPdsBase("https://")); +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"); -- tangled.sh