diff --git a/examples/hackernews/README.md b/examples/hackernews/README.md index 574c9c6cd6..20e70682ed 100644 --- a/examples/hackernews/README.md +++ b/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. diff --git a/examples/hackernews/qualification.json b/examples/hackernews/qualification.json index ed62d24657..45aecaef6a 100644 --- a/examples/hackernews/qualification.json +++ b/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." + ] } } diff --git a/zig/pkg/antfly-embedded/src/index.zig b/zig/pkg/antfly-embedded/src/index.zig index f015009462..ea83cee02f 100644 --- a/zig/pkg/antfly-embedded/src/index.zig +++ b/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); +} diff --git a/zig/pkg/antfly-embedded/src/metadata/lake_index_catalog.zig b/zig/pkg/antfly-embedded/src/metadata/lake_index_catalog.zig index c37f73481a..ef2afdcd11 100644 --- a/zig/pkg/antfly-embedded/src/metadata/lake_index_catalog.zig +++ b/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; diff --git a/zig/pkg/antfly-embedded/src/section/inverted.zig b/zig/pkg/antfly-embedded/src/section/inverted.zig index cec8c27e09..eedd366d8f 100644 --- a/zig/pkg/antfly-embedded/src/section/inverted.zig +++ b/zig/pkg/antfly-embedded/src/section/inverted.zig @@ -8867,6 +8867,14 @@ pub const RangeInvertedIndexReader = struct { if (dict_length < term_dict_header_size or dict_length > view.length - v7_header_size) return error.InvalidData; const dict_offset = view.length - dict_length; if (@as(u64, bloom_length) + norms_length > dict_offset - v7_header_size) return error.InvalidData; + // The header identifies three independent required metadata reads. + // Hint them together so remote sources overlap their bounded requests; + // required reads still authenticate bytes and surface every error. + view.source.prefetch(view.offset + dict_offset, term_dict_header_size); + if (bloom_length >= bloom.magic.len + 13) + view.source.prefetch(view.offset + dict_offset - bloom_length, bloom.magic.len + 13); + if (norms_length >= 5) + view.source.prefetch(view.offset + dict_offset - bloom_length - norms_length, 5); var dict: [term_dict_header_size]u8 = undefined; try view.readInto(dict_offset, &dict); if (!std.mem.eql(u8, dict[0..4], term_dict_magic)) return error.InvalidData; diff --git a/zig/pkg/antfly-embedded/src/serverless/query/lake_parquet_rowgroup.zig b/zig/pkg/antfly-embedded/src/serverless/query/lake_parquet_rowgroup.zig index 5897eaff37..d838a7df1a 100644 --- a/zig/pkg/antfly-embedded/src/serverless/query/lake_parquet_rowgroup.zig +++ b/zig/pkg/antfly-embedded/src/serverless/query/lake_parquet_rowgroup.zig @@ -1013,9 +1013,10 @@ const PersistentObjectRangeCacheState = struct { const manager = self.resource_manager orelse return null; const source = manager.capacitySource() orelse return null; const observation = try source.current(); - const timestamp_ns = std.Io.Timestamp.now(self.io, .awake).toNanoseconds(); - const now_ns: u64 = std.math.cast(u64, timestamp_ns) orelse - if (timestamp_ns < 0) 0 else std.math.maxInt(u64); + // Capacity sources timestamp observations in the native process clock. + // Io awake has a different origin on Darwin (UPTIME_RAW), so comparing + // it with CLOCK_MONOTONIC can reject every freshly observed write. + const now_ns = @import("antfly_platform").time.monotonicNs(); return manager.reserveCapacity( self.alloc, source.domain_id, @@ -6707,6 +6708,47 @@ test "lake persistent object range cache bounds write-behind admission" { try std.testing.expectEqual(@as(usize, 1), persistent.statsSnapshot().read_misses); } +test "lake persistent cache admits native-clock capacity with an offset Io awake clock" { + const a = std.testing.allocator; + var io_impl = std.Io.Threaded.init(a, .{}); + defer io_impl.deinit(); + const native_io = io_impl.io(); + var shifted_vtable = native_io.vtable.*; + const Clock = struct { + fn now(_: ?*anyopaque, clock: std.Io.Clock) std.Io.Timestamp { + if (clock == .awake) return .{ .nanoseconds = 1 }; + return std.testing.io.vtable.now(std.testing.io.userdata, clock); + } + fn observe(_: *anyopaque) !resource_manager_mod.CapacityObservation { + return .{ .available_bytes = 1024 * 1024 * 1024 * 1024, .capacity_bytes = 1024 * 1024 * 1024 * 1024, .observed_at_ns = @import("antfly_platform").time.monotonicNs(), .valid_for_ns = 5 * std.time.ns_per_s }; + } + }; + shifted_vtable.now = Clock.now; + const io: std.Io = .{ .userdata = native_io.userdata, .vtable = &shifted_vtable }; + var manager = resource_manager_mod.ResourceManager.init(.{}); + defer manager.deinit(a); + var tag: u8 = 0; + try manager.installCapacitySource(.{ .ptr = &tag, .domain_id = 7, .observe = Clock.observe }); + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const root = try std.fmt.allocPrint(a, ".zig-cache/tmp/{s}/clock-cache", .{tmp.sub_path}); + defer a.free(root); + { + var cache = try PersistentObjectRangeCache.initWithPolicyAndResources(io, root, .{}, .{ .resource_manager = &manager }); + defer cache.deinit(); + try std.testing.expectEqual(.enqueued, cache.enqueueWrite("key", "data")); + cache.flush(); + try std.testing.expectEqual(@as(usize, 1), cache.statsSnapshot().writes_completed); + try std.testing.expectEqual(@as(usize, 0), cache.statsSnapshot().writes_dropped); + } + var restarted = try PersistentObjectRangeCache.init(io, root); + defer restarted.deinit(); + const bytes = (try restarted.readAlloc(a, "key", 4)).?; + defer a.free(bytes); + try std.testing.expectEqualStrings("data", bytes); + try std.testing.expectEqual(@as(usize, 1), restarted.statsSnapshot().read_hits); +} + test "lake persistent object range cache evicts least recently used entries within disk ceilings" { var io_impl = std.Io.Threaded.init(std.testing.allocator, .{}); defer io_impl.deinit(); diff --git a/zig/pkg/antfly-embedded/src/serverless/query/lake_serving_cache.zig b/zig/pkg/antfly-embedded/src/serverless/query/lake_serving_cache.zig index 6e2cb9eb46..d92b37a3a1 100644 --- a/zig/pkg/antfly-embedded/src/serverless/query/lake_serving_cache.zig +++ b/zig/pkg/antfly-embedded/src/serverless/query/lake_serving_cache.zig @@ -79,6 +79,7 @@ pub const Cache = struct { while (!self.mutex.tryLock()) std.atomic.spinLoopHint(); self.stats.disk_unavailable = @errorName(err); self.mutex.unlock(); + std.log.scoped(.lake_cache).warn("persistent cache unavailable; serving through RAM/source: {s}", .{@errorName(err)}); self.persistent_attempted = true; return; }; diff --git a/zig/pkg/antfly/src/api/lake_index_coordinator.zig b/zig/pkg/antfly/src/api/lake_index_coordinator.zig index c340d147b0..6122358682 100644 --- a/zig/pkg/antfly/src/api/lake_index_coordinator.zig +++ b/zig/pkg/antfly/src/api/lake_index_coordinator.zig @@ -283,7 +283,7 @@ test "external lake native coordinator fences ambiguous admission and reuses dur try std.testing.expect(try durableDirectoryAvailable(a, &store, recovered.value.published.?, .none)); // A complete older directory is still a refresh obligation after a // reader-format upgrade, even when every source signature is unchanged. - recovered.value.published.?.reader_protocol = 28; + recovered.value.published.?.reader_protocol = 31; const historical = try catalog.encode(a, recovered.value); if (mock.owned) |bytes| a.free(bytes); mock.owned = historical; diff --git a/zig/pkg/antfly/src/api/lake_index_native_text_cache.zig b/zig/pkg/antfly/src/api/lake_index_native_text_cache.zig index 270056fe1f..1940402bbf 100644 --- a/zig/pkg/antfly/src/api/lake_index_native_text_cache.zig +++ b/zig/pkg/antfly/src/api/lake_index_native_text_cache.zig @@ -371,31 +371,29 @@ const Entry = struct { const replacements = try self.allocator.alloc(local.index.ReplacementSegmentData, additions.items.len); defer self.allocator.free(replacements); var loaded: usize = 0; - defer for (replacements[0..loaded]) |*replacement| replacement.data.deinit(self.allocator); + defer for (replacements[0..loaded]) |*replacement| deinitReplacement(replacement, self.allocator); var loader: corpus.CachedSegments = .{ .store = store, .cache = cached, .seekable = root.seekable, .query_owned = true, .resource_manager = self.resource_manager }; var read_bytes: u64 = 512 * 1024 * 1024; var position: usize = 0; while (position < additions.items.len) { const end = @min(position + 4, additions.items.len); - var tasks: [4]?local.sql_parallel_scheduler.Task(anyerror!local.index.SegmentData) = @splat(null); + var tasks: [4]?local.sql_parallel_scheduler.Task(anyerror!local.index.ReplacementSegmentData) = @splat(null); defer for (&tasks) |*task| if (task.*) |*pending| if (pending.future != null) { - var data = pending.cancel(io) catch continue; - data.deinit(self.allocator); + var replacement = pending.cancel(io) catch continue; + deinitReplacement(&replacement, self.allocator); }; for (position..end) |ordinal| { const ref = additions.items[ordinal]; try cancellation.check(); try stores.chargeReadBudget(&read_bytes, ref.byte_len); - tasks[ordinal - position] = local.sql_parallel_scheduler.global().submit(io, @intCast(ref.byte_len), loadSegment, .{ &loader, self.allocator, ref, cancellation }); + tasks[ordinal - position] = local.sql_parallel_scheduler.global().submit(io, @intCast(ref.byte_len), loadSegment, .{ &loader, self.allocator, ref, cancellation, ids.items[ordinal] }); if (tasks[ordinal - position] == null) { - const data = try loadSegment(&loader, self.allocator, ref, cancellation); - replacements[loaded] = .{ .id = ids.items[ordinal], .data = data }; + replacements[loaded] = try loadSegment(&loader, self.allocator, ref, cancellation, ids.items[ordinal]); loaded += 1; } } for (position..end) |ordinal| if (tasks[ordinal - position]) |*pending| { - const data = try pending.await(io); - replacements[loaded] = .{ .id = ids.items[ordinal], .data = data }; + replacements[loaded] = try pending.await(io); loaded += 1; }; position = end; @@ -416,9 +414,21 @@ const Entry = struct { try self.writer.?.orderImmutableSegments(ordered_ids); if (self.resource_manager) |manager| self.writer.?.attachResourceManager(manager); } - fn loadSegment(loader: *corpus.CachedSegments, a: A, ref: artifacts.ChunkRef, cancellation: Cancellation) anyerror!local.index.SegmentData { + fn loadSegment(loader: *corpus.CachedSegments, a: A, ref: artifacts.ChunkRef, cancellation: Cancellation, id: u64) anyerror!local.index.ReplacementSegmentData { const mapped = loader.loader(); - return mapped.load(mapped.ptr, a, ref, cancellation); + var data = try mapped.load(mapped.ptr, a, ref, cancellation); + errdefer data.deinit(a); + // Directory loading alone leaves footer/field metadata I/O until the + // writer's serial publication loop. Admit those required reads in the + // same bounded jobs, then transfer each reader without reopening it. + var reader = try data.initReader(a); + errdefer reader.deinit(); + try cancellation.check(); + return .{ .id = id, .data = data, .prepared_reader = reader }; + } + fn deinitReplacement(replacement: *local.index.ReplacementSegmentData, a: A) void { + if (replacement.prepared_reader) |*reader| reader.deinit(); + replacement.data.deinit(a); } fn awaitReady(self: *Entry, io: std.Io, context: Context) !void { while (self.state.load(.acquire) == .loading) { diff --git a/zig/pkg/antfly/src/api/sql_pgwire.zig b/zig/pkg/antfly/src/api/sql_pgwire.zig index bfdace2ad8..492038fbb7 100644 --- a/zig/pkg/antfly/src/api/sql_pgwire.zig +++ b/zig/pkg/antfly/src/api/sql_pgwire.zig @@ -969,9 +969,14 @@ const Authority = struct { fn context(self: *Authority) !operation.RequestContext { try self.checkpoint(); try self.credential.validate(); - return .{ - .deadline_ns = std.math.cast(u64, self.request.deadline.raw.nanoseconds) orelse return error.DeadlineExceeded, - .deadline_io = io_abi.Borrow.init(&self.request.io), + // Normalize once at ingress before catalog/storage owner boundaries, + // which carry native absolute deadlines rather than an Io clock borrow. + // Unrepresentable positive deadlines are effectively unbounded; check() + // above still rejects expired requests and explicit cancellation. + const deadline = std.math.cast(u64, self.request.deadline.raw.nanoseconds); + const native_context: operation.RequestContext = .{ + .deadline_ns = deadline, + .deadline_io = if (deadline != null) io_abi.Borrow.init(&self.request.io) else null, .fanout_io = io_abi.Borrow.init(&self.request.io), .cancellation = operation.CancellationToken.fromAtomic(self.request.cancel_requested), .principal = .{ .kind = .user, .subject = self.identity.*.?.username }, @@ -982,6 +987,7 @@ const Authority = struct { .row_policy_credential = self.identity, .table_write_authorization = .{ .ptr = self, .allows = allowsWrite }, }; + return native_context.platformDeadline(); } fn allowsWrite(raw: *const anyopaque, table: []const u8) bool { @@ -1418,6 +1424,35 @@ test "SQL pgwire credential snapshot observes policy revocation and password rot .deadline = .{ .clock = .awake, .raw = .{ .nanoseconds = std.math.maxInt(i96) } }, .cancel_requested = &cancellation, } }; + { + const saved = authority.request; + defer authority.request = saved; + const OffsetClock = struct { + fn now(_: ?*anyopaque, clock: std.Io.Clock) std.Io.Timestamp { + if (clock == .awake) return .{ .nanoseconds = std.time.ns_per_s }; + return std.testing.io.vtable.now(std.testing.io.userdata, clock); + } + }; + var vtable = std.testing.io.vtable.*; + vtable.now = OffsetClock.now; + authority.request.io = .{ .userdata = std.testing.io.userdata, .vtable = &vtable }; + authority.request.deadline.raw.nanoseconds = 10 * std.time.ns_per_s; + const before = @import("antfly_platform").time.monotonicNs(); + const normalized = try authority.context(); + const after = @import("antfly_platform").time.monotonicNs(); + try std.testing.expect(normalized.deadline_io == null); + try std.testing.expect(normalized.deadline_ns.? >= before + 9 * std.time.ns_per_s); + try std.testing.expect(normalized.deadline_ns.? <= after + 9 * std.time.ns_per_s); + try normalized.ensureActive(); + authority.request.deadline.raw.nanoseconds = std.math.maxInt(i96); + try std.testing.expect((try authority.context()).deadline_ns == null); + authority.request.deadline.raw.nanoseconds = std.time.ns_per_s; + try std.testing.expectError(error.QueryCanceled, authority.context()); + authority.request.deadline.raw.nanoseconds = 10 * std.time.ns_per_s; + cancellation.store(true, .release); + defer cancellation.store(false, .release); + try std.testing.expectError(error.QueryCanceled, authority.context()); + } const revision: ?u64 = 1; var guarded: GuardedCatalog = .{ .native = .{ .ptr = &native_cursor, .vtable = &.{ .resolve = undefined, .scan = undefined, .mutate = undefined, .open_scan = NativeCursor.open, .checkpoint = NativeCursor.checkpoint } }, diff --git a/zig/pkg/antfly/src/serverless/artifacts/object_store.zig b/zig/pkg/antfly/src/serverless/artifacts/object_store.zig index ff40c95be5..90c9d6c347 100644 --- a/zig/pkg/antfly/src/serverless/artifacts/object_store.zig +++ b/zig/pkg/antfly/src/serverless/artifacts/object_store.zig @@ -330,6 +330,35 @@ pub const ObjectStore = struct { return try dupeWithCancellationAlloc(alloc, result.body, cancellation); } + fn getBoundedAllocWithCancellation( + self: *ObjectStore, + alloc: std.mem.Allocator, + artifact_id: []const u8, + expected_len: usize, + cancellation: CancellationToken, + ) ![]u8 { + try cancellation.check(); + const key = try keyForArtifactIdAlloc(self.alloc, self.prefix, artifact_id); + defer self.alloc.free(key); + var client = self.client; + client.allocator = alloc; + // Fetch the complete body: a prefix Range GET cannot prove object length. + // The content-addressed identity is authenticated by the caller's hash. + var result = client.getObject(self.bucket, key, .{ + .skip_metadata_probe = true, + .max_response_bytes = @max(expected_len, 1), + .cancellation = objectstore.CancellationToken.fromCallback(cancellation.ptr, cancellation.is_cancelled_fn), + }) catch |err| { + if (err == error.ResponseTooLarge) return error.ArtifactIntegrityMismatch; + return normalizeCancellationError(err, cancellation); + }; + defer result.deinit(alloc); + try cancellation.check(); + const body = result.body; + result.body = &.{}; + return body; + } + pub fn getRangeAlloc(self: *ObjectStore, alloc: std.mem.Allocator, artifact_id: []const u8, offset: u64, len: usize) ![]u8 { return try self.getRangeAllocWithCancellation(alloc, artifact_id, offset, len, .none); } @@ -668,6 +697,7 @@ pub const ObjectStore = struct { .reclaim_retired_scoped_inventory = erasedReclaimRetiredScopedInventory, .get_alloc = erasedGetAlloc, .get_alloc_with_cancellation = erasedGetAllocWithCancellation, + .get_bounded_alloc_with_cancellation = erasedGetBoundedAllocWithCancellation, .get_range_alloc = erasedGetRangeAlloc, .get_range_alloc_with_cancellation = erasedGetRangeAllocWithCancellation, .get_verified_range_alloc_with_cancellation = erasedGetVerifiedRangeAllocWithCancellation, @@ -722,6 +752,11 @@ pub const ObjectStore = struct { return try self.getAllocWithCancellation(alloc, artifact_id, cancellation); } + fn erasedGetBoundedAllocWithCancellation(ptr: *anyopaque, alloc: std.mem.Allocator, artifact_id: []const u8, expected_len: usize, cancellation: CancellationToken) ![]u8 { + const self: *ObjectStore = @ptrCast(@alignCast(ptr)); + return self.getBoundedAllocWithCancellation(alloc, artifact_id, expected_len, cancellation); + } + fn erasedGetRangeAlloc(ptr: *anyopaque, alloc: std.mem.Allocator, artifact_id: []const u8, offset: u64, len: usize) ![]u8 { const self: *ObjectStore = @ptrCast(@alignCast(ptr)); return try self.getRangeAlloc(alloc, artifact_id, offset, len); @@ -1180,3 +1215,34 @@ fn cleanupTmp(path: [*:0]const u8) void { defer io_impl.deinit(); std.Io.Dir.cwd().deleteTree(io_impl.io(), std.mem.span(path)) catch {}; } + +test "external lake complete artifact reads avoid metadata probes and reject tampering" { + const alloc = std.testing.allocator; + var memory = objectstore.MemoryClient.init(alloc); + defer memory.deinit(); + var impl = try ObjectStore.initWithClient(alloc, memory.client(), "bucket", "cold"); + var store = impl.artifactStore(); + defer store.deinit(); + var metadata = try store.put("verified"); + defer metadata.deinit(alloc); + memory.resetOperationCount(); + const body = try store.getVerifiedAllocWithCancellation(metadata.artifact_id, metadata.byte_len, metadata.checksum, .none); + defer alloc.free(body); + try std.testing.expectEqualStrings("verified", body); + try std.testing.expectEqual(@as(u64, 1), memory.operationCount()); + const key = try keyForArtifactIdAlloc(alloc, impl.prefix, metadata.artifact_id); + defer alloc.free(key); + var client = memory.client(); + for ([_][]const u8{ "verified-extra", "short", "modified" }) |corrupt| { + var put = try client.putObject("bucket", key, corrupt, .{}); + put.deinit(alloc); + try std.testing.expectError(error.ArtifactIntegrityMismatch, store.getVerifiedAllocWithCancellation(metadata.artifact_id, metadata.byte_len, metadata.checksum, .none)); + } + var empty = try store.put(""); + defer empty.deinit(alloc); + const empty_body = try store.getVerifiedAllocWithCancellation(empty.artifact_id, 0, empty.checksum, .none); + defer alloc.free(empty_body); + try std.testing.expectEqual(@as(usize, 0), empty_body.len); + try store.delete(empty.artifact_id); + try std.testing.expectError(error.FileNotFound, store.getVerifiedAllocWithCancellation(empty.artifact_id, 0, empty.checksum, .none)); +} diff --git a/zig/pkg/antfly/src/serverless/artifacts/store.zig b/zig/pkg/antfly/src/serverless/artifacts/store.zig index b7a1bcbf4a..3024bdb17b 100644 --- a/zig/pkg/antfly/src/serverless/artifacts/store.zig +++ b/zig/pkg/antfly/src/serverless/artifacts/store.zig @@ -199,6 +199,10 @@ pub const ArtifactStore = struct { put_with_cancellation: ?*const fn (*anyopaque, Allocator, []const u8, CancellationToken) anyerror!ArtifactMetadata = null, get_alloc: *const fn (*anyopaque, Allocator, []const u8) anyerror![]u8, get_alloc_with_cancellation: ?*const fn (*anyopaque, Allocator, []const u8, CancellationToken) anyerror![]u8 = null, + /// Return the complete object without a metadata probe, bounded by the + /// expected size. The caller verifies exact length and SHA-256; a range + /// read cannot substitute because it would hide appended bytes. + get_bounded_alloc_with_cancellation: ?*const fn (*anyopaque, Allocator, []const u8, usize, CancellationToken) anyerror![]u8 = null, get_range_alloc: *const fn (*anyopaque, Allocator, []const u8, u64, usize) anyerror![]u8, get_range_alloc_with_cancellation: ?*const fn (*anyopaque, Allocator, []const u8, u64, usize, CancellationToken) anyerror![]u8 = null, get_verified_range_alloc_with_cancellation: ?*const fn (*anyopaque, Allocator, []const u8, u64, []const u8, u64, usize, CancellationToken) anyerror![]u8 = null, @@ -341,6 +345,14 @@ pub const ArtifactStore = struct { const expected_len = std.math.cast(usize, expected_byte_len) orelse return error.ArtifactTooLarge; + if (self.vtable.get_bounded_alloc_with_cancellation) |get_bounded| { + const payload = try get_bounded(self.ptr, result_alloc, artifact_id, expected_len, cancellation); + errdefer result_alloc.free(payload); + if (payload.len != expected_len) return error.ArtifactIntegrityMismatch; + try validatePayloadSha256WithCancellation(payload, expected_checksum, cancellation); + return payload; + } + { var metadata = try self.statWithCancellationUsingAllocator(result_alloc, artifact_id, cancellation); defer metadata.deinit(result_alloc);