Fix lake cold startup, restart cache persistence, and pgwire deadlines

examples/hackernews/README.md

@@ -67,28 +67,43 @@ connections have separate prefix allowlists and capabilities (`lake_read` and
 | Normalized, allocator fix only | 26.4 s | 17.8 s | 27.5 s | Rejected; unordered timed out |
 | Same normalized file, follow-up fixes | 22.3 s | 468 ms | 20.0 s | Passed |
 | Untouched BigQuery export, follow-up fixes | 13.7 s | 452 ms | 15.3 s | Passed |
+| Untouched export, cold/cache fixes, two empty-cache runs | 7.6–10.2 s | 379–400 ms | 418–432 ms | Passed |
 
 Search previously cached inventory metadata but never attached the Parquet range
-cache for hydration. Attaching it removes repeated remote reads on warm queries.
-First-query and restart latency still need work: the configured disk-cache
-directory contained no persisted range files. The first query follows index
-publication, so it is not a guaranteed cold-cache measurement. Readiness timing
-on an existing publication is not index-build timing. These results do not
-establish full-archive capacity or concurrent production latency.
+cache for hydration. Attaching it removed repeated remote reads on warm queries.
+The subsequent cold/cache fixes remove an index-wide dictionary diagnostic scan
+from range-backed reader admission, admit readers in bounded parallel jobs, and
+fetch complete immutable artifacts with one bounded, checksum-verified GET.
+Metadata hints overlap independent required header reads.
+
+A clock-domain mismatch prevented disk-cache writes on macOS: capacity probes
+use the native monotonic clock, while admission used Io's awake clock. Cache
+admission now uses the probe's clock. Both new runs persisted 82 cache files
+(about 5 MiB) and reused them after a real process restart. Pgwire normalizes its
+awake-clock deadline before native catalog/storage boundaries; the six prior
+cancellation cases now pass without longer timeouts.
+
+The two new runs copied catalog data into a new state directory with no cache
+files before startup. The harness SQL count precedes text search, so Parquet
+metadata can already be warm. The text index was cold. Older first-query timings
+followed publication. Readiness timing on an existing publication is not
+index-build timing. Empty-cache queries still pay required remote I/O. These
+results do not establish full-archive capacity or concurrent production latency.
 
 A separate raw-export check also passed the previously timed-out unordered
 HN-ID filter in 481 ms.
 
-Flat structured predicates now scan only their required columns against the
-pinned, delete-aware lake snapshot and resolve IDs before ranking/pagination.
-This preserves exact integer comparisons and works when predicate fields are
-absent from the result projection or text index. Match sets are bounded at
-100,000 IDs; nested paths retain the existing residual-filter path. Full-archive
-filtering needs indexed predicate evaluation rather than a full-source scan.
+Tables with declared relational indexes evaluate supported structured predicates
+against indexed physical candidates before ranking, including match sets above
+100,000 rows. This example currently declares only its full-text index: its
+flat structured predicates scan projected columns against the pinned,
+delete-aware lake snapshot and collect up to 100,000 matching IDs. Full-archive
+qualification needs relational indexes on the HN metadata fields and larger
+partitions. Nested paths retain the existing residual-filter path.
 
 ## Remaining deployment work
 
-- Diagnose persistent cache initialization/writes and reduce cold GCS read fanout.
+- Measure remaining cold remote I/O in the deployment region and under concurrency.
 - Qualify larger partitions against build/corpus limits and indexed filtering.
 - Define durable buckets and service identities through Colony's infra workflow.
 - Stream backfills, resolve parent stories, and reconcile edits/deletions.

examples/hackernews/qualification.json

@@ -63,9 +63,9 @@
   },
   "limits": [
     "Local Debug build, 10000-row smoke check against GCS, not a production benchmark.",
-    "Initial query follows publication and is not a guaranteed cold-cache measurement.",
-    "First query after restart still takes 15\u201320 seconds; persistent range-cache files were not observed.",
-    "Flat structured filters scan projected columns and collect up to 100000 matching IDs; full-archive filtering needs indexed predicate evaluation.",
+    "Historical first-query measurements followed publication. New cold runs started with no disk cache; SQL count still precedes text search.",
+    "Empty-cache search still pays required remote metadata, posting and hydration I/O: 7.6–10.2 seconds locally in the new runs.",
+    "The example declares only a full-text index, so its metadata filters use a projected-column scan. Full-archive filtering requires declared relational indexes.",
     "Staging bucket expires all objects after eight days."
   ],
   "unordered_filter_regression": {
@@ -74,5 +74,66 @@
     "unordered_filter_ms": 480.61,
     "warm_search_ms": 522.3,
     "after_restart_ms": 15515.21
+  },
+  "cold_start_and_cancellation_fixes": {
+    "runtime_base_commit": "6b92267b7d",
+    "source": "gs://colony-import-sources-antfly-dev-01/hn-poc/20261007/archive",
+    "row_count": 10000,
+    "cache_setup": "Copied catalog only, no cache directory before server startup. The harness SQL count precedes the first text search.",
+    "main_upgrade_baseline": {
+      "first_search_ms": 16523.45,
+      "warm_search_ms": 368.89,
+      "after_restart_ms": 18125.06,
+      "persisted_cache_files": 0
+    },
+    "runs": [
+      {
+        "first_search_ms": 7603.29,
+        "warm_search_ms": 399.94,
+        "after_restart_ms": 418.03,
+        "unordered_filter_ms": 379.73,
+        "ready_seconds": 11.81,
+        "verified": [
+          "remote_bm25",
+          "projected_highlights",
+          "snapshot_pagination",
+          "restart_scores",
+          "restart_pagination",
+          "exact_filter"
+        ]
+      },
+      {
+        "first_search_ms": 10164.46,
+        "warm_search_ms": 378.79,
+        "after_restart_ms": 432.21,
+        "unordered_filter_ms": 405.06,
+        "ready_seconds": 0.79,
+        "verified": [
+          "remote_bm25",
+          "projected_highlights",
+          "snapshot_pagination",
+          "restart_scores",
+          "restart_pagination",
+          "exact_filter"
+        ]
+      }
+    ],
+    "persisted_cache_files": 82,
+    "persisted_cache_bytes": 5133368,
+    "verification": {
+      "lake_tests_passed": 695,
+      "pgwire_clock_regressions_passed": 1,
+      "previous_pgwire_cancellations_passed": 6,
+      "lake_sql_e2e_passed": 20
+    },
+    "changes": [
+      "Native-clock cache capacity admission permits writes and restart reuse.",
+      "Range-backed index admission defers dictionary diagnostics.",
+      "Bounded parallel reader admission transfers prepared readers without reopening.",
+      "Required inverted metadata reads use bounded overlapping prefetch.",
+      "Complete immutable artifacts use one bounded GET with length and SHA-256 verification.",
+      "Pgwire normalizes Io deadlines before native owner boundaries.",
+      "Catalog reader protocols 24 through 33 remain readable for reconciliation."
+    ]
   }
 }

zig/pkg/antfly-embedded/src/index.zig

@@ -571,7 +571,10 @@ pub const SegmentEntry = struct {
     id: u64,
     data: SegmentData,
     reader: segment_mod.SegmentReader,
-    layout_stats: segment_mod.SegmentLayoutStats = .{},
+    // Range-backed readers compute diagnostics only when requested. Counting
+    // dictionary terms at admission would turn a sparse cold query into a
+    // scan of every term block in every segment.
+    layout_stats: ?segment_mod.SegmentLayoutStats = null,
     shared: *SegmentShared,
     // Query-bound native readers borrow the same physical segment pin but own
     // their decoder/navigation state and capability-bound source separately.
@@ -775,14 +778,24 @@ pub const SegmentEntry = struct {
     }
 
     pub fn layoutStats(self: *const SegmentEntry, detailed_inverted: bool) segment_mod.SegmentLayoutStats {
-        if (!detailed_inverted) return self.layout_stats;
+        if (!detailed_inverted) return self.layout_stats orelse self.reader.layoutStats();
         return self.reader.layoutStatsWithInvertedDetails(true);
     }
+
+    fn admissionLayoutStats(data: SegmentData, reader: *const SegmentReader) ?segment_mod.SegmentLayoutStats {
+        if (data == .native and data.native == .ranges) return null;
+        return reader.layoutStats();
+    }
 };
 
 pub const ReplacementSegmentData = struct {
     id: u64,
     data: SegmentData,
+    /// An admitted reader created from `data` with the writer allocator.
+    /// Preparation consumes and clears this reader even if it later fails;
+    /// data ownership still transfers only on publication. This lets remote
+    /// owners admit metadata in bounded parallel jobs before the writer lock.
+    prepared_reader: ?SegmentReader = null,
     /// Optional tombstones to install in the same snapshot publication as
     /// the replacement data. The caller retains ownership; IndexWriter clones
     /// the bitmap into the replacement SegmentShared cell.
@@ -2080,7 +2093,7 @@ pub const IndexWriter = struct {
             .id = seg_id,
             .data = data,
             .reader = reader,
-            .layout_stats = reader.layoutStats(),
+            .layout_stats = SegmentEntry.admissionLayoutStats(data, &reader),
             .shared = shared,
         };
 
@@ -2254,7 +2267,7 @@ pub const IndexWriter = struct {
             .id = seg_id,
             .data = owned.?,
             .reader = reader,
-            .layout_stats = reader.layoutStats(),
+            .layout_stats = SegmentEntry.admissionLayoutStats(owned.?, &reader),
             .shared = shared,
         };
 
@@ -2354,7 +2367,10 @@ pub const IndexWriter = struct {
         }
 
         for (replacements, 0..) |*replacement, i| {
-            replacement_readers[i] = try replacement.data.initReader(self.alloc);
+            replacement_readers[i] = if (replacement.prepared_reader) |reader| blk: {
+                replacement.prepared_reader = null;
+                break :blk reader;
+            } else try replacement.data.initReader(self.alloc);
             if (replacement.data == .owned_view) replacement_readers[i].postings_loader = replacement.data.owned_view.postings_loader;
             replacement_readers_initialized += 1;
         }
@@ -2423,7 +2439,7 @@ pub const IndexWriter = struct {
                 .id = replacement.id,
                 .data = replacement.data,
                 .reader = replacement_readers[i],
-                .layout_stats = replacement_readers[i].layoutStats(),
+                .layout_stats = SegmentEntry.admissionLayoutStats(replacement.data, &replacement_readers[i]),
                 .shared = shared,
             };
             cells_created += 1;
@@ -3913,3 +3929,78 @@ test "external lake scoring parallel lanes join before releasing query authority
     try std.testing.expectEqual(@as(usize, 0), scheduler.workers);
     try std.testing.expectEqual(@as(usize, 0), scheduler.bytes);
 }
+
+test "external lake range segment admission defers dictionary diagnostics across replacements" {
+    const a = std.testing.allocator;
+    var text = inverted.InvertedIndexBuilder.init(a, .{});
+    defer text.deinit();
+    var segment_writer = segment_mod.SegmentWriter.init(a);
+    defer segment_writer.deinit();
+    for (0..16384) |doc| {
+        var term: [32]u8 = undefined;
+        try text.addDocument(@intCast(doc), &.{.{ .term = try std.fmt.bufPrint(&term, "term-{d:0>8}", .{doc}), .freq = 1, .positions = &.{0} }});
+        try segment_writer.addUnstoredDoc();
+    }
+    const section = try text.build();
+    defer a.free(section);
+    try segment_writer.addSection(try segment_writer.addField("body"), .inverted_text, section);
+    const bytes = try segment_writer.build();
+    defer a.free(bytes);
+    var contiguous = try SegmentReader.init(a, bytes);
+    defer contiguous.deinit();
+    const expected = contiguous.layoutStats();
+    const State = struct {
+        bytes: []const u8,
+        reads: usize = 0,
+        fn read(raw: *anyopaque, offset: u64, out: []u8) !void {
+            const self: *@This() = @ptrCast(@alignCast(raw));
+            self.reads += 1;
+            @memcpy(out, self.bytes[@intCast(offset)..][0..out.len]);
+        }
+        fn close(_: *anyopaque) void {}
+        fn data(self: *@This()) SegmentData {
+            return .fromNative(.{ .ranges = .{ .ptr = self, .length = self.bytes.len, .read_into = read, .close = close } });
+        }
+    };
+    var state = State{ .bytes = bytes };
+    var writer = try IndexWriter.init(a);
+    defer writer.deinit();
+    try writer.addSegmentWithIdData(1, state.data());
+    var entry = &writer.snapshot().segments[0];
+    // Opening must stay bounded independently of dictionary term/block count.
+    try std.testing.expect(state.reads < 32);
+    const admitted_reads = state.reads;
+    try std.testing.expectEqualDeep(expected, entry.layoutStats(false));
+    try std.testing.expect(state.reads > admitted_reads);
+    state.reads = 0;
+    var replacements = [_]ReplacementSegmentData{.{ .id = 2, .data = state.data() }};
+    try writer.replaceSegmentsManyData(&.{1}, &replacements);
+    entry = &writer.snapshot().segments[0];
+    try std.testing.expectEqual(@as(u64, 2), entry.id);
+    try std.testing.expect(state.reads < 32);
+    try std.testing.expectEqualDeep(expected, entry.layoutStats(false));
+    try std.testing.expectEqual(@as(u64, 16384), entry.layoutStats(false).inverted_term_count);
+    var admitted = [_]ReplacementSegmentData{.{ .id = 3, .data = state.data(), .prepared_reader = try state.data().initReader(a) }};
+    defer if (admitted[0].prepared_reader) |*reader| reader.deinit();
+    const before_adoption = state.reads;
+    try writer.replaceSegmentsManyData(&.{2}, &admitted);
+    try std.testing.expect(admitted[0].prepared_reader == null);
+    try std.testing.expectEqual(before_adoption, state.reads);
+    try std.testing.expectEqual(@as(u64, 3), writer.snapshot().segments[0].id);
+    // Fail after reader transfer but before allocating the replacement snapshot.
+    // Preparation owns reader cleanup; the unpublished data remains ours.
+    var unpublished = [_]ReplacementSegmentData{.{ .id = 4, .data = state.data(), .prepared_reader = try state.data().initReader(a) }};
+    defer unpublished[0].data.deinit(a);
+    defer if (unpublished[0].prepared_reader) |*reader| reader.deinit();
+    var failing = std.testing.FailingAllocator.init(a, .{ .fail_index = 0 });
+    writer.alloc = failing.allocator();
+    const attempted = writer.prepareSegmentsManyDataWithScratch(a, &.{3}, &unpublished);
+    writer.alloc = a;
+    if (attempted) |value| {
+        var unexpected = value;
+        unexpected.abort();
+        return error.TestExpectedError;
+    } else |err| try std.testing.expectEqual(error.OutOfMemory, err);
+    try std.testing.expect(unpublished[0].prepared_reader == null);
+    try std.testing.expectEqual(@as(u64, 3), writer.snapshot().segments[0].id);
+}

zig/pkg/antfly-embedded/src/metadata/lake_index_catalog.zig

@@ -27,6 +27,12 @@ pub const Token = [16]u8;
 pub const DirectoryRef = struct { artifact_id: []const u8, checksum: []const u8, byte_len: u64, count: u32 };
 pub const max_directory_artifacts: usize = 4096;
 pub const native_reader_protocol: u16 = 33;
+
+// Catalog envelopes must remain readable so reconciliation can replace older
+// derived formats. Serving still requires the current reader protocol.
+fn validateReaderProtocol(protocol: u16) !void {
+    if (protocol != 0 and (protocol < 24 or protocol > native_reader_protocol)) return error.InvalidLakeIndexCatalog;
+}
 pub const max_contributions: usize = 1024 * 1024;
 pub const max_directory_bytes: usize = 16 * 1024 * 1024;
 /// Physical namespace plus a named connection for current credential lookup.
@@ -64,7 +70,7 @@ pub const Attempt = struct {
     started_at_ms: u64,
     lease_expires_at_ms: u64,
     fn validate(self: Attempt) !void {
-        if (self.reader_protocol != 0 and self.reader_protocol != 24 and self.reader_protocol != 25 and self.reader_protocol != 26 and self.reader_protocol != 27 and self.reader_protocol != 28 and self.reader_protocol != 29 and self.reader_protocol != native_reader_protocol) return error.InvalidLakeIndexCatalog;
+        try validateReaderProtocol(self.reader_protocol);
         if (self.store_locator) |locator| try locator.validate();
         try self.signature.validate();
         if (self.generation == 0 or std.mem.allEqual(u8, &self.token, 0) or self.lease_expires_at_ms <= self.started_at_ms) return error.InvalidLakeIndexCatalog;
@@ -109,7 +115,7 @@ pub const Publication = struct {
         }
     }
     pub fn validate(self: Publication) !void {
-        if (self.reader_protocol != 0 and self.reader_protocol != 24 and self.reader_protocol != 25 and self.reader_protocol != 26 and self.reader_protocol != 27 and self.reader_protocol != 28 and self.reader_protocol != 29 and self.reader_protocol != native_reader_protocol) return error.InvalidLakeIndexCatalog;
+        try validateReaderProtocol(self.reader_protocol);
         if (self.store_locator) |locator| try locator.validate();
         if (self.namespace) |namespace| if (std.mem.allEqual(u8, &namespace, 0)) return error.InvalidLakeIndexCatalog;
         if (self.generation == 0 or std.mem.allEqual(u8, &self.token, 0) or self.declarations.len > (if (self.directory != null) max_directory_artifacts else max_artifacts)) return error.InvalidLakeIndexCatalog;
@@ -364,6 +370,32 @@ test "metadata.lake index publication preserves ready roots and fences changed d
         .inventory = inventory,
         .declarations = &.{},
     };
+    // A reader upgrade must parse both historical pending attempts and ready
+    // publications before it can rebuild them under the current protocol.
+    for (24..native_reader_protocol + 1) |protocol| {
+        var historical_attempt = attempt;
+        historical_attempt.pending.?.reader_protocol = @intCast(protocol);
+        const pending_bytes = try encode(a, historical_attempt);
+        defer a.free(pending_bytes);
+        var pending = try parse(a, pending_bytes);
+        defer pending.deinit();
+        try std.testing.expectEqual(protocol, pending.value.pending.?.reader_protocol);
+        var historical_publication = publication;
+        historical_publication.reader_protocol = @intCast(protocol);
+        const published_bytes = try encode(a, .{ .generation = 1, .published = historical_publication });
+        defer a.free(published_bytes);
+        var published_history = try parse(a, published_bytes);
+        defer published_history.deinit();
+        try std.testing.expectEqual(protocol, published_history.value.published.?.reader_protocol);
+    }
+    for ([_]u16{ 23, native_reader_protocol + 1 }) |protocol| {
+        var invalid_attempt = attempt.pending.?;
+        invalid_attempt.reader_protocol = protocol;
+        try std.testing.expectError(error.InvalidLakeIndexCatalog, invalid_attempt.validate());
+        var invalid_reader = publication;
+        invalid_reader.reader_protocol = protocol;
+        try std.testing.expectError(error.InvalidLakeIndexCatalog, invalid_reader.validate());
+    }
     const building_bytes = try encode(a, attempt);
     defer a.free(building_bytes);
     var building = table;