From 3669540f2a86b42dc69370e513d46ad412837292 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sat, 15 Aug 2026 17:21:46 -0500 Subject: [PATCH] retry: PDS-direct-first getRepo with relay fallback (upstream retry.go parity) the failed-repo retry pass routed every getRepo through the relay's 302, concentrating the whole pass on bsky.network from the firehose's own IP (suspect in the 08-14 4h ingest stall). upstream's retryRunner.download dials the stamped PDS first and uses the relay only when the stamp is unroutable or the stamped host authoritatively lacks the repo; fallback success restamps the route. ported, with upstream's foreign-embedded-DID guard for the direct path, two counters, and regression tests that pin the stale-stamp 404 -> relay fallback -> restamp path (without which a migrated repo would be marked terminal-complete undownloaded). Co-Authored-By: Claude Fable 5 --- HANDOFF.md | 21 ++- src/internal/ingest/backfill/resync.zig | 232 +++++++++++++++++++++++- src/internal/ingest/backfill/retry.zig | 6 +- src/internal/runtime/metrics.zig | 12 ++ 4 files changed, 261 insertions(+), 10 deletions(-) diff --git a/HANDOFF.md b/HANDOFF.md index 3917177..bd2f7a7 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -27,13 +27,20 @@ closed and chronicled in git history 08-11..08-12. decides: per-IP shaping -> pace retry + port PDS-direct backfill; both-dip -> network-wide; neither -> theory dies. Mitigation proposed to operator: retry interval 24h + host-workers 1, compaction 24h — - APPLIED 08-15 ~05:55Z (config-only restart). THE FIX, found upstream: - their retry fetches the KNOWN PDS first and uses the relay 302 only as - fallback (retry.go:425-438); our resync.zig routes everything through - bsky.network. Port PDS-direct-first (task #11) — parity, not invention — - then consider restoring the 4h interval. Shaping theory stays unproven - (mitigation restart reset the retry timer, cancelling the experiment); - the mitigation covers it either way. + APPLIED 08-15 ~05:09Z (config-only restart; earlier "~05:55Z" was a + wrong clock — the durable last-pass gauge boot-stamps at 05:09:46Z). + THE FIX, found upstream: their retry fetches the KNOWN PDS first and + uses the relay 302 only as fallback (retry.go:425-438); our resync.zig + routed everything through bsky.network. PORTED (task #11): resyncRepo + now dials the stamped `pds` first; unroutable stamp or a direct 404 + falls back to the relay and restamps on success; foreign embedded DID + from the direct path is rejected (GetRepoDidMismatch). Counters: + stream_backfill_failed_repo_retry_{pds_direct,relay_fallback}_total. + Regression tests: stale-stamp fallback + restamp, DID mismatch, in + resync.zig. After a clean retry pass with direct routing, consider + restoring the 4h interval. Shaping theory stays unproven (mitigation + restart reset the retry timer, cancelling the experiment); the port + removes the relay-hammering pattern either way. - Prior deploy `56dbf2e` (receipt 21/21; 401 contract byte-identical to Bluesky's edge, probed side-by-side). Prod archive IS key-gated: STREAM_ARCHIVE_API_KEY in /opt/stream-experiment/.env (never in git), diff --git a/src/internal/ingest/backfill/resync.zig b/src/internal/ingest/backfill/resync.zig index 16ca22b..e204759 100644 --- a/src/internal/ingest/backfill/resync.zig +++ b/src/internal/ingest/backfill/resync.zig @@ -35,6 +35,10 @@ 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. @@ -51,6 +55,7 @@ pub fn resyncRepo( io: Io, scratch_dir: []const u8, relay_http: []const u8, + pds: []const u8, did: []const u8, archive: *archive_mod.Archive, store: *repo_store.Store, @@ -68,10 +73,26 @@ pub fn resyncRepo( }; defer if (previous) |state| store.freeState(allocator, state); const active = if (previous) |state| state.active else true; - const outcome = prepared_fetch.fetch(allocator, alloc, io, scratch_dir, relay_http, did, &details.final_host) catch |err| { + // 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. + 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| { 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| { @@ -103,6 +124,13 @@ pub fn resyncRepo( }, }; 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{ @@ -162,7 +190,12 @@ pub fn resyncRepo( .latest_rev = head_rev, .host = details.final_host.slice(), .handle = if (previous) |state| state.handle else "", - .pds = if (previous) |state| state.pds 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) + 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, @@ -177,6 +210,18 @@ pub fn resyncRepo( 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; +} + fn terminalState(previous: ?repo_store.RepoState, status: repo_store.Status, active: bool, host: []const u8, terminal_us: i64) repo_store.RepoState { return .{ .status = status, @@ -306,6 +351,189 @@ test "formatRfc3339" { 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://")); +} + +const meta_store = @import("../../storage/meta_store.zig"); + +/// full signed single-record repo CAR, same fixture shape as repos.zig tests +fn 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, .{secret} ** 32); + 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(); diff --git a/src/internal/ingest/backfill/retry.zig b/src/internal/ingest/backfill/retry.zig index 51f7ec0..7a8936b 100644 --- a/src/internal/ingest/backfill/retry.zig +++ b/src/internal/ingest/backfill/retry.zig @@ -447,7 +447,7 @@ const RetryWork = struct { var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/repair-scratch/failed-repo", .{self.retry.data_dir}); if (self.retry.stats) |s| _ = s.retry_attempts_total.fetchAdd(1, .monotonic); - _ = resync.resyncRepo(self.retry.allocator, self.retry.io, scratch_dir, self.retry.relay_http, did, self.retry.archive, self.store, &drops, &details, self.retry.trace_context) catch |err| { + _ = resync.resyncRepo(self.retry.allocator, self.retry.io, scratch_dir, self.retry.relay_http, state.pds, did, self.retry.archive, self.store, &drops, &details, self.retry.trace_context) catch |err| { if (err == error.InvalidCommitRev) { if (self.retry.stats) |s| _ = s.backfill_drops.getPtr(.invalid_rev).fetchAdd(1, .monotonic); } @@ -499,6 +499,10 @@ const RetryWork = struct { _ = self.succeeded.fetchAdd(1, .monotonic); if (self.retry.stats) |s| { _ = s.retry_resynced_total.fetchAdd(1, .monotonic); + if (details.via_fallback) + _ = s.retry_relay_fallback_total.fetchAdd(1, .monotonic) + else + _ = s.retry_pds_direct_total.fetchAdd(1, .monotonic); _ = s.backfill_drops.getPtr(.invalid_collection).fetchAdd(drops.invalid_collection, .monotonic); _ = s.backfill_drops.getPtr(.invalid_rkey).fetchAdd(drops.invalid_rkey, .monotonic); _ = s.backfill_drops.getPtr(.field_too_long).fetchAdd(drops.field_too_long, .monotonic); diff --git a/src/internal/runtime/metrics.zig b/src/internal/runtime/metrics.zig index 5b56c52..e005826 100644 --- a/src/internal/runtime/metrics.zig +++ b/src/internal/runtime/metrics.zig @@ -211,6 +211,8 @@ pub const Stats = struct { retry_candidates_total: Counter = .init(0), retry_attempts_total: Counter = .init(0), retry_skipped_host_parked_total: Counter = .init(0), + retry_pds_direct_total: Counter = .init(0), + retry_relay_fallback_total: Counter = .init(0), backfill_workers_active: std.atomic.Value(usize) = .init(0), backfill_queue_depth: std.atomic.Value(usize) = .init(0), import_jobs_ok_total: Counter = .init(0), @@ -1182,6 +1184,10 @@ pub fn format( \\jetstream_backfill_failed_repo_retry_failed_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_skipped_host_parked_total counter \\jetstream_backfill_failed_repo_retry_skipped_host_parked_total {d} + \\# TYPE stream_backfill_failed_repo_retry_pds_direct_total counter + \\stream_backfill_failed_repo_retry_pds_direct_total {d} + \\# TYPE stream_backfill_failed_repo_retry_relay_fallback_total counter + \\stream_backfill_failed_repo_retry_relay_fallback_total {d} \\# TYPE stream_backfill_repos_total counter \\stream_backfill_repos_total{{result="complete"}} {d} \\stream_backfill_repos_total{{result="failed"}} {d} @@ -1212,6 +1218,8 @@ pub fn format( stats.retry_resynced_total.load(.monotonic), stats.retry_failed_total.load(.monotonic), stats.retry_skipped_host_parked_total.load(.monotonic), + stats.retry_pds_direct_total.load(.monotonic), + stats.retry_relay_fallback_total.load(.monotonic), stats.backfill_completed_total.load(.monotonic), stats.backfill_failed_total.load(.monotonic), stats.backfill_workers_active.load(.monotonic), @@ -1575,6 +1583,8 @@ test "format renders counters" { _ = stats.retry_passes_total.fetchAdd(8, .monotonic); _ = stats.retry_candidates_total.fetchAdd(9, .monotonic); _ = stats.retry_skipped_host_parked_total.fetchAdd(10, .monotonic); + _ = stats.retry_pds_direct_total.fetchAdd(11, .monotonic); + _ = stats.retry_relay_fallback_total.fetchAdd(12, .monotonic); stats.backfill_workers_active.store(4, .monotonic); stats.backfill_queue_depth.store(23, .monotonic); _ = stats.import_jobs_ok_total.fetchAdd(1, .monotonic); @@ -1736,6 +1746,8 @@ test "format renders counters" { try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_passes_total 8") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_candidates_total 9") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_skipped_host_parked_total 10") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_failed_repo_retry_pds_direct_total 11") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_failed_repo_retry_relay_fallback_total 12") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_blocks_flushed_total 6") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_segments_rotated_total 7") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_append_errors_total 8") != null); -- 2.51.2