Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335//! gloom v0.1.0 blocked bloom filter — bit-exact zig port//!//! the gloom wire format (jcalabro/gloom, pinned upstream) is part of the//! jss segment format: DID blooms in sealed footers are gloom MarshalBinary//! blobs, and their no-false-negative contract requires our Add to place//! exactly the same bits as the Go implementation. every formula and//! constant here mirrors gloom's params.go/bloom.go/hash.go; the parameter//! derivation is float-sensitive, so tests pin golden vectors generated//! from the Go library.
const std = @import("std");const xxhash128 = @import("xxhash128.zig");
const Allocator = std.mem.Allocator;
pub const block_bits = 512;pub const block_words = 8;const min_k = 3;const max_k = 17;const ln2_squared = 0.4804530139182014;const blocking_growth_cap = 4.0;const header_size = 21;
/// prime partitions per k; each row sums to 512 (gloom params.go:40-56)const prime_partitions = [_][]const u32{ &.{}, &.{}, &.{}, // k 0..2 unused &.{ 167, 173, 172 }, &.{ 109, 127, 137, 139 }, &.{ 97, 101, 103, 109, 102 }, &.{ 61, 79, 83, 89, 97, 103 }, &.{ 61, 67, 71, 79, 83, 89, 62 }, &.{ 37, 47, 53, 61, 67, 71, 79, 97 }, &.{ 41, 43, 47, 53, 59, 67, 71, 73, 58 }, &.{ 31, 37, 41, 43, 47, 53, 59, 61, 67, 73 }, &.{ 29, 31, 37, 41, 43, 44, 47, 53, 59, 61, 67 }, &.{ 17, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 71 }, &.{ 17, 19, 23, 29, 31, 37, 41, 43, 47, 52, 53, 59, 61 }, &.{ 11, 13, 17, 19, 23, 29, 31, 37, 41, 47, 53, 59, 61, 71 }, &.{ 11, 13, 17, 19, 23, 28, 29, 31, 37, 41, 43, 47, 53, 59, 61 }, &.{ 5, 7, 11, 13, 17, 19, 23, 27, 29, 31, 37, 41, 43, 47, 53, 109 }, &.{ 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 32, 37, 41, 43, 47, 53, 101 },};
pub const Params = struct { num_blocks: u64, k: u32 };
/// gloom OptimalParams (params.go:84-117)pub fn optimalParams(expected_items: u64, fp_rate: f64) Params { var n = expected_items; if (n == 0) n = 1; var p = fp_rate; if (p <= 0) p = 0.0001; if (p >= 1) p = 0.99;
const bits_per_item = -@log(p) / ln2_squared; const total_bits = @as(f64, @floatFromInt(n)) * bits_per_item; const base_blocks: u64 = @intFromFloat(@ceil(total_bits / block_bits)); const max_blocks: u64 = @intFromFloat(@ceil(@as(f64, @floatFromInt(base_blocks)) * blocking_growth_cap));
var num_blocks = base_blocks; var k = bestK(num_blocks, n); while (estimateFalsePositiveRate(num_blocks, k, n) > p and num_blocks < max_blocks) { num_blocks = @min(num_blocks + num_blocks / 8 + 1, max_blocks); k = bestK(num_blocks, n); } return .{ .num_blocks = num_blocks, .k = k };}
fn bestK(num_blocks: u64, items: u64) u32 { var best: u32 = min_k; var best_fp = std.math.inf(f64); var k: u32 = min_k; while (k <= max_k) : (k += 1) { const fp = estimateFalsePositiveRate(num_blocks, k, items); if (fp < best_fp) { best_fp = fp; best = k; } } return best;}
fn partitionedBlockFP(primes: []const u32, j: f64) f64 { var fp: f64 = 1.0; for (primes) |prime| { const p_f: f64 = @floatFromInt(prime); fp *= 1.0 - std.math.pow(f64, 1.0 - 1.0 / p_f, j); } return fp;}
fn estimateFalsePositiveRate(num_blocks: u64, k: u32, items_added: u64) f64 { if (num_blocks == 0 or items_added == 0) return 0; const primes = prime_partitions[k]; const lambda = @as(f64, @floatFromInt(items_added)) / @as(f64, @floatFromInt(num_blocks)); if (lambda > 10000) return partitionedBlockFP(primes, lambda);
const max_j: i64 = @intFromFloat(lambda + 10 * @sqrt(lambda) + 20); var fp: f64 = 0; var log_factorial: f64 = 0; var j: i64 = 0; while (j <= max_j) : (j += 1) { const j_f: f64 = @floatFromInt(j); if (j > 0) log_factorial += @log(j_f); const log_prob = -lambda + j_f * @log(lambda) - log_factorial; const prob = @exp(log_prob); if (prob < 1e-15 and j > @as(i64, @intFromFloat(lambda))) break; if (j > 0) fp += prob * partitionedBlockFP(primes, j_f); } return fp;}
pub const Filter = struct { num_blocks: u64, k: u32, count: u64 = 0, words: []u64, // num_blocks × 8, zeroed
pub fn init(allocator: Allocator, expected_items: u64, fp_rate: f64) !Filter { const params = optimalParams(expected_items, fp_rate); return initWithParams(allocator, params.num_blocks, params.k); }
pub fn initWithParams(allocator: Allocator, num_blocks_in: u64, k_in: u32) !Filter { const num_blocks = if (num_blocks_in == 0) 1 else num_blocks_in; const k = if (k_in < min_k or k_in > max_k) 7 else k_in; const words = try allocator.alloc(u64, num_blocks * block_words); @memset(words, 0); return .{ .num_blocks = num_blocks, .k = k, .words = words }; }
pub fn deinit(self: *Filter, allocator: Allocator) void { allocator.free(self.words); }
pub fn add(self: *Filter, key: []const u8) void { const h = xxhash128.hash(key); const block_idx = mulhi64(h.hi, self.num_blocks); const intra = foldIntra(h.lo); const primes = prime_partitions[self.k]; var offset: u32 = 0; for (primes) |prime| { const bit = offset + intra % prime; self.words[block_idx * block_words + bit / 64] |= @as(u64, 1) << @intCast(bit % 64); offset += prime; } self.count += 1; }
pub fn contains(self: *const Filter, key: []const u8) bool { const h = xxhash128.hash(key); const block_idx = mulhi64(h.hi, self.num_blocks); const intra = foldIntra(h.lo); const primes = prime_partitions[self.k]; var offset: u32 = 0; for (primes) |prime| { const bit = offset + intra % prime; if (self.words[block_idx * block_words + bit / 64] & (@as(u64, 1) << @intCast(bit % 64)) == 0) return false; offset += prime; } return true; }
/// gloom MarshalBinary: version|k|numBlocks|count|words|crc32c, all LE pub fn marshal(self: *const Filter, allocator: Allocator) ![]u8 { const total = header_size + self.words.len * 8 + 4; const out = try allocator.alloc(u8, total); out[0] = 1; std.mem.writeInt(u32, out[1..5], self.k, .little); std.mem.writeInt(u64, out[5..13], self.num_blocks, .little); std.mem.writeInt(u64, out[13..21], self.count, .little); for (self.words, 0..) |w, i| { std.mem.writeInt(u64, out[header_size + i * 8 ..][0..8], w, .little); } const crc = std.hash.crc.@"CRC-32/ISCSI".hash(out[0 .. total - 4]); std.mem.writeInt(u32, out[total - 4 ..][0..4], crc, .little); return out; }
/// serialized size for these dimensions (all per-block blooms share it) pub fn marshaledSize(self: *const Filter) usize { return header_size + self.words.len * 8 + 4; }};
fn mulhi64(a: u64, b: u64) u64 { return @truncate((@as(u128, a) * @as(u128, b)) >> 64);}
fn foldIntra(lo: u64) u32 { return @as(u32, @truncate(lo)) ^ @as(u32, @truncate(lo >> 32));}
// === tests ===// golden vectors generated from Go gloom v0.1.0 (the pinned upstream dep)
const testing = std.testing;
test "optimalParams matches Go goldens (p=0.001)" { const goldens = [_]struct { n: u64, num_blocks: u64, k: u32 }{ .{ .n = 1, .num_blocks = 1, .k = 15 }, .{ .n = 2, .num_blocks = 1, .k = 15 }, .{ .n = 10, .num_blocks = 1, .k = 15 }, .{ .n = 50, .num_blocks = 2, .k = 11 }, .{ .n = 100, .num_blocks = 4, .k = 11 }, .{ .n = 1000, .num_blocks = 33, .k = 9 }, .{ .n = 4096, .num_blocks = 131, .k = 9 }, .{ .n = 10000, .num_blocks = 317, .k = 9 }, .{ .n = 100000, .num_blocks = 3161, .k = 9 }, }; for (goldens) |g| { const params = optimalParams(g.n, 0.001); try testing.expectEqual(g.num_blocks, params.num_blocks); try testing.expectEqual(g.k, params.k); }}
test "marshal matches Go byte-for-byte" { var f = try Filter.init(testing.allocator, 10, 0.001); defer f.deinit(testing.allocator); f.add("did:plc:streamsampleaccount"); f.add("did:plc:another"); const blob = try f.marshal(testing.allocator); defer testing.allocator.free(blob);
const golden_hex = "010f000000010000000000000002000000000000000444104000c0000080000a8000004100001000800000100000100000201000080008000000000400080000000200000008000004000000040000000000000048ced03a4e"; var golden: [89]u8 = undefined; _ = try std.fmt.hexToBytes(&golden, golden_hex); try testing.expectEqualSlices(u8, &golden, blob);}
test "membership: added keys always found" { var f = try Filter.init(testing.allocator, 100, 0.001); defer f.deinit(testing.allocator); var buf: [32]u8 = undefined; for (0..100) |i| { const key = try std.fmt.bufPrint(&buf, "did:plc:account{d}", .{i}); f.add(key); } for (0..100) |i| { const key = try std.fmt.bufPrint(&buf, "did:plc:account{d}", .{i}); try testing.expect(f.contains(key)); } try testing.expectEqual(@as(u64, 100), f.count); try testing.expect(!f.contains("did:plc:streamsampleaccount"));}
/// parse a MarshalBinary blob (crc-checked). the returned filter borrows/// nothing — words are copied.pub fn unmarshal(allocator: Allocator, blob: []const u8) !Filter { if (blob.len < header_size + 4) return error.Truncated; const crc_stored = std.mem.readInt(u32, blob[blob.len - 4 ..][0..4], .little); if (std.hash.crc.@"CRC-32/ISCSI".hash(blob[0 .. blob.len - 4]) != crc_stored) return error.BadChecksum; if (blob[0] != 1) return error.UnsupportedVersion; const k = std.mem.readInt(u32, blob[1..5], .little); if (k < min_k or k > max_k) return error.UnsupportedK; const num_blocks = std.mem.readInt(u64, blob[5..13], .little); if (num_blocks == 0 or num_blocks > 1 << 50) return error.BadBlockCount; const count = std.mem.readInt(u64, blob[13..21], .little); const want_len = header_size + num_blocks * block_words * 8 + 4; if (blob.len != want_len) return error.LengthMismatch;
const words = try allocator.alloc(u64, num_blocks * block_words); errdefer allocator.free(words); for (words, 0..) |*w, i| { w.* = std.mem.readInt(u64, blob[header_size + i * 8 ..][0..8], .little); } return .{ .num_blocks = num_blocks, .k = k, .count = count, .words = words };}
/// Validate a serialized resident filter once when manifest metadata loads.pub fn validateMarshaled(blob: []const u8) !void { if (blob.len < header_size + 4) return error.Truncated; const crc_stored = std.mem.readInt(u32, blob[blob.len - 4 ..][0..4], .little); if (std.hash.crc.@"CRC-32/ISCSI".hash(blob[0 .. blob.len - 4]) != crc_stored) return error.BadChecksum; _ = try dimensions(blob);}
const Dimensions = struct { k: u32, num_blocks: u64 };
fn dimensions(blob: []const u8) !Dimensions { if (blob.len < header_size + 4 or blob[0] != 1) return error.Truncated; const k = std.mem.readInt(u32, blob[1..5], .little); if (k < min_k or k > max_k) return error.UnsupportedK; const num_blocks = std.mem.readInt(u64, blob[5..13], .little); if (num_blocks == 0 or num_blocks > 1 << 50) return error.BadBlockCount; const want_len = header_size + num_blocks * block_words * 8 + 4; if (blob.len != want_len) return error.LengthMismatch; return .{ .k = k, .num_blocks = num_blocks };}
/// Test a key directly against a resident, load-validated gloom blob./// This avoids allocating/copying every filter on every planSnapshot request.pub fn containsMarshaled(blob: []const u8, key: []const u8) !bool { const dims = try dimensions(blob); const k = dims.k; const num_blocks = dims.num_blocks;
const h = xxhash128.hash(key); const block_idx = mulhi64(h.hi, num_blocks); const intra = foldIntra(h.lo); var partition_offset: u32 = 0; for (prime_partitions[k]) |prime| { const bit = partition_offset + intra % prime; const word_index = block_idx * block_words + bit / 64; const at: usize = @intCast(header_size + word_index * 8); const word = std.mem.readInt(u64, blob[at..][0..8], .little); if (word & (@as(u64, 1) << @intCast(bit % 64)) == 0) return false; partition_offset += prime; } return true;}
test "marshal/unmarshal round-trip preserves membership" { var f = try Filter.init(testing.allocator, 50, 0.001); defer f.deinit(testing.allocator); f.add("did:plc:one"); f.add("did:plc:two"); const blob = try f.marshal(testing.allocator); defer testing.allocator.free(blob);
var g = try unmarshal(testing.allocator, blob); defer g.deinit(testing.allocator); try testing.expect(g.contains("did:plc:one")); try testing.expect(g.contains("did:plc:two")); try testing.expect(!g.contains("did:plc:three")); try testing.expectEqual(f.count, g.count);
// corrupt a byte → crc rejects const bad = try testing.allocator.dupe(u8, blob); defer testing.allocator.free(bad); bad[header_size] ^= 1; try testing.expectError(error.BadChecksum, unmarshal(testing.allocator, bad));}
test "resident marshaled membership matches decoded filter" { var f = try Filter.init(testing.allocator, 50, 0.001); defer f.deinit(testing.allocator); f.add("did:plc:one"); const blob = try f.marshal(testing.allocator); defer testing.allocator.free(blob); try testing.expect(try containsMarshaled(blob, "did:plc:one")); try testing.expect(!try containsMarshaled(blob, "did:plc:absent"));}