feat(video): add bounded packet history, delayed FIFO and owned snapshots

packages/video/src/core/packet_ring.zigadded+191-0

Changes to packages/video/src/core/packet_ring.zig

		@@ -0,0 +1,191 @@
	1	+//! Owned encoded-packet history. One caller serializes all access. Fixed byte
	2	+//! arena and entry array; push never allocates, blocks, or keeps borrowed input.
	3	+//! Eviction drops dependent packets until the oldest retained packet is a key.
	4	+const std = ;
	5	+pub const Entry = struct { offset: usize, length: usize, timestamp: u64, key: bool };
	6	+pub const Ring = struct {
	7	+ allocator: std.mem.Allocator,
	8	+ bytes: []u8,
	9	+ entries: []Entry,
	10	+ duration: u64,
	11	+ head: usize = 0,
	12	+ count: usize = 0,
	13	+ used: usize = 0,
	14	+ write_pos: usize = 0,
	15	+ last: ?u64 = null,
	16	+ evicted: u64 = 0,
	17	+ skipped: u64 = 0,
	18	+ pub fn init(allocator: std.mem.Allocator, byte_capacity: usize, frame_capacity: usize, duration: u64) !Ring {
	19	+ if (byte_capacity == 0 or frame_capacity == 0 or frame_capacity > std.math.maxInt(usize) / 2 or duration == 0) return error.InvalidConfiguration;
	20	+ const bytes = try allocator.alloc(u8, byte_capacity);
	21	+ errdefer allocator.free(bytes);
	22	+ const entries = try allocator.alloc(Entry, frame_capacity);
	23	+ return .{ .allocator = allocator, .bytes = bytes, .entries = entries, .duration = duration };
	24	+ }
	25	+ pub fn deinit(s: *Ring) void {
	26	+ s.allocator.free(s.bytes);
	27	+ s.allocator.free(s.entries);
	28	+ }
	29	+ fn drop(s: *Ring) void {
	30	+ s.used -= s.entries[s.head].length;
	31	+ s.head = (s.head + 1) % s.entries.len;
	32	+ s.count -= 1;
	33	+ s.evicted += 1;
	34	+ }
	35	+ pub fn entry(s: *const Ring, index: usize) ?Entry {
	36	+ if (index >= s.count) return null;
	37	+ return s.entries[(s.head + index) % s.entries.len];
	38	+ }
	39	+ /// False means waiting for a key after eviction. Limits are upper bounds;
	40	+ /// history can be empty when a GOP is larger than the retention limits.
	41	+ /// Oversized/invalid input is rejected before modifying retained history.
	42	+ pub fn push(s: *Ring, packet: []const u8, timestamp: u64, key: bool) !bool {
	43	+ if (packet.len == 0 or packet.len > s.bytes.len) return error.PacketTooLarge;
	44	+ if (s.last) |last| if (timestamp <= last) return error.InvalidTimestamp;
	45	+ while (s.count > 0 and (s.count == s.entries.len or packet.len > s.bytes.len - s.used or timestamp - s.entry(0).?.timestamp > s.duration)) s.drop();
	46	+ while (s.count > 0 and !s.entry(0).?.key) s.drop();
	47	+ s.last = timestamp;
	48	+ if (s.count == 0 and !key) {
	49	+ s.skipped += 1;
	50	+ return false;
	51	+ }
	52	+ s.store(packet, timestamp, key);
	53	+ return true;
	54	+ }
	55	+ fn store(s: *Ring, packet: []const u8, timestamp: u64, key: bool) void {
	56	+ s.last = timestamp;
	57	+ const first = @min(packet.len, s.bytes.len - s.write_pos);
	58	+ @memcpy(s.bytes[s.write_pos..][0..first], packet[0..first]);
	59	+ @memcpy(s.bytes[0 .. packet.len - first], packet[first..]);
	60	+ s.entries[(s.head + s.count) % s.entries.len] = .{ .offset = s.write_pos, .length = packet.len, .timestamp = timestamp, .key = key };
	61	+ s.write_pos = if (packet.len - first != 0) packet.len - first else if (s.write_pos + first == s.bytes.len) 0 else s.write_pos + first;
	62	+ s.count += 1;
	63	+ s.used += packet.len;
	64	+ }
	65	+ /// FIFO mode: no implicit eviction or GOP filtering. A consumer must pop
	66	+ /// mature entries before enqueueing; overflow leaves all state unchanged.
	67	+ pub fn pushQueued(s: *Ring, packet: []const u8, timestamp: u64, key: bool) !void {
	68	+ if (packet.len == 0 or packet.len > s.bytes.len) return error.PacketTooLarge;
	69	+ if (s.last) |last| if (timestamp <= last) return error.InvalidTimestamp;
	70	+ if (s.count == s.entries.len or packet.len > s.bytes.len - s.used) return error.BufferFull;
	71	+ s.store(packet, timestamp, key);
	72	+ }
	73	+ pub fn pop(s: *Ring) void {
	74	+ if (s.count != 0) s.drop();
	75	+ }
	76	+ /// Null clock flushes EOF; subtraction avoids overflowing timestamp+delay.
	77	+ pub fn ready(s: *const Ring, now: ?u64, delay: u64) bool {
	78	+ const e = s.entry(0) orelse return false;
	79	+ const clock = now orelse return true;
	80	+ return clock >= e.timestamp and clock - e.timestamp >= delay;
	81	+ }
	82	+ /// Own only the selected suffix. Borrowed source data never escapes.
	83	+ pub fn cloneFrom(s: *const Ring, a: std.mem.Allocator, start: usize) !Ring {
	84	+ if (start >= s.count) return error.InvalidIndex;
	85	+ var bytes: usize = 0;
	86	+ for (start..s.count) |i| bytes = try std.math.add(usize, bytes, s.entry(i).?.length);
	87	+ var result = try Ring.init(a, bytes, s.count - start, s.duration);
	88	+ errdefer result.deinit();
	89	+ // Destination is contiguous in this clone; copy straight into its arena.
	90	+ for (start..s.count) |i| {
	91	+ const e = s.entry(i).?;
	92	+ _ = try s.copy(i, result.bytes[result.used..]);
	93	+ result.entries[result.count] = .{ .offset = result.used, .length = e.length, .timestamp = e.timestamp, .key = e.key };
	94	+ result.count += 1;
	95	+ result.used += e.length;
	96	+ result.last = e.timestamp;
	97	+ }
	98	+ result.write_pos = 0;
	99	+ return result;
	100	+ }
	101	+ /// Explicit owned snapshot copy; data and metadata views otherwise expire
	102	+ /// on the next mutation. Returns the active destination slice.
	103	+ pub fn copy(s: *const Ring, index: usize, out: []u8) ![]const u8 {
	104	+ const e = s.entry(index) orelse return error.InvalidIndex;
	105	+ if (out.len < e.length) return error.BufferTooSmall;
	106	+ const first = @min(e.length, s.bytes.len - e.offset);
	107	+ @memcpy(out[0..first], s.bytes[e.offset..][0..first]);
	108	+ @memcpy(out[first..e.length], s.bytes[0 .. e.length - first]);
	109	+ return out[0..e.length];
	110	+ }
	111	+};
	112	+test "byte wrap and GOP eviction keep only independently decodable history" {
	113	+ var ring = try Ring.init(std.testing.allocator, 10, 4, 100);
	114	+ defer ring.deinit();
	115	+ try std.testing.expect(try ring.push("aaaa", 0, true));
	116	+ try std.testing.expect(try ring.push("bb", 1, false));
	117	+ try std.testing.expect(try ring.push("ccc", 2, true));
	118	+ try std.testing.expect(try ring.push("dddd", 3, false));
	119	+ try std.testing.expectEqual(@as(usize, 2), ring.count);
	120	+ try std.testing.expectEqual(@as(usize, 7), ring.used);
	121	+ var out: [10]u8 = undefined;
	122	+ try std.testing.expectEqualStrings("ccc", try ring.copy(0, &out));
	123	+ try std.testing.expectEqualStrings("dddd", try ring.copy(1, &out));
	124	+ try std.testing.expect(!try ring.push("12345678", 4, false));
	125	+ try std.testing.expectEqual(@as(usize, 0), ring.count);
	126	+ try std.testing.expect(try ring.push("key", 5, true));
	127	+ try std.testing.expectEqualStrings("key", try ring.copy(0, &out));
	128	+}
	129	+test "frame and time limits; rejected inputs preserve existing payload" {
	130	+ var ring = try Ring.init(std.testing.allocator, 10, 2, 3);
	131	+ defer ring.deinit();
	132	+ try std.testing.expect(!try ring.push("p", 0, false));
	133	+ try std.testing.expect(try ring.push("a", 1, true));
	134	+ try std.testing.expect(try ring.push("b", 2, false));
	135	+ try std.testing.expect(!try ring.push("c", 3, false));
	136	+ try std.testing.expect(try ring.push("d", 4, true));
	137	+ try std.testing.expectError(error.PacketTooLarge, ring.push("01234567890", 5, true));
	138	+ try std.testing.expectError(error.InvalidTimestamp, ring.push("x", 4, true));
	139	+ var out: [1]u8 = undefined;
	140	+ try std.testing.expectEqualStrings("d", try ring.copy(0, &out));
	141	+ try std.testing.expect(!try ring.push("e", 8, false));
	142	+ try std.testing.expectEqual(@as(usize, 0), ring.count);
	143	+ try std.testing.expectError(error.InvalidIndex, ring.copy(0, &out));
	144	+}
	145	+test "initialization rejects zero limits and frees partial allocations" {
	146	+ try std.testing.expectError(error.InvalidConfiguration, Ring.init(std.testing.allocator, 0, 1, 1));
	147	+ const Check = struct {
	148	+ fn run(a: std.mem.Allocator) !void {
	149	+ var r = try Ring.init(a, 10, 2, 3);
	150	+ defer r.deinit();
	151	+ }
	152	+ };
	153	+ try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
	154	+}
	155	+
	156	+test "FIFO delay preserves dependencies and rejects overflow atomically" {
	157	+ var r = try Ring.init(std.testing.allocator, 8, 3, 1);
	158	+ defer r.deinit();
	159	+ try r.pushQueued("key", 10, true);
	160	+ try r.pushQueued("p", 20, false);
	161	+ try std.testing.expect(!r.ready(19, 10));
	162	+ try std.testing.expect(r.ready(20, 10));
	163	+ r.pop();
	164	+ try std.testing.expect(!r.entry(0).?.key);
	165	+ try r.pushQueued("next", 30, false);
	166	+ try std.testing.expectError(error.BufferFull, r.pushQueued("full", 40, true));
	167	+ try std.testing.expectEqual(@as(u64, 30), r.last.?);
	168	+ try std.testing.expectEqual(@as(usize, 2), r.count);
	169	+ try std.testing.expect(r.ready(null, std.math.maxInt(u64)));
	170	+ try std.testing.expect(!r.ready(0, 0));
	171	+ var clone = try r.cloneFrom(std.testing.allocator, 0);
	172	+ defer clone.deinit();
	173	+ r.pop();
	174	+ r.pop();
	175	+ var out: [8]u8 = undefined;
	176	+ try std.testing.expectEqualStrings("next", try clone.copy(1, &out));
	177	+}
	178	+test "suffix cloning frees partial allocations and preserves source" {
	179	+ const Check = struct {
	180	+ fn run(a: std.mem.Allocator) !void {
	181	+ var r = try Ring.init(a, 9, 3, 10);
	182	+ defer r.deinit();
	183	+ _ = try r.push("first", 0, true);
	184	+ _ = try r.push("next", 1, true);
	185	+ var frozen = try r.cloneFrom(a, 1);
	186	+ defer frozen.deinit();
	187	+ try std.testing.expectEqual(@as(usize, 2), r.count);
	188	+ }
	189	+ };
	190	+ try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
	191	+}