ReviewOS
Reload

also looking at this

home-lang/home

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

#916
Merged chris wants to merge feat/keyframe-packet-ring into main
3 files +229 -0

Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.

packages/video/docs/packet-ring.mdmodified+12-0
@@ -22,3 +22,15 @@ zig-camera uses this module with Home JPEG/Matroska and x264/x265/fragmented MP4
2222It retains compressed access units, without retaining camera DMA leases or raw
2323RGB frames. Explicit snapshots need scratch space for one largest retained
2424packet; remuxing starts with fresh codec headers and resets file timestamps.
25
26For time-shifted streaming, `pushQueued` uses the same owned arena as a bounded
27FIFO: it never evicts or filters dependent frames, and rejects full/oversized or
28non-monotonic input without mutation. Use `ready(now, delay)` then `copy(0, scratch)`
29and `pop()` to forward matured packets; `ready(null, delay)` flushes EOF. Queue
30capacity must cover the delay. The caller controls keyframe starts and output
31errors; a full FIFO must not silently discard encoded dependencies.
32
33`cloneFrom(allocator, start)` owns a minimally sized copy of a selected suffix.
34Serialize it with source mutation; the resulting data can move to a worker and
35outlive the source. It copies wrapped bytes directly into a contiguous arena and
36cleans up partial allocation failures. The module creates no threads or file I/O.
packages/video/src/core/packet_ring.zigmodified+78-1
@@ -49,6 +49,11 @@ pub const Ring = struct {
4949 s.skipped += 1;
5050 return false;
5151 }
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;
5257 const first = @min(packet.len, s.bytes.len - s.write_pos);
5358 @memcpy(s.bytes[s.write_pos..][0..first], packet[0..first]);
5459 @memcpy(s.bytes[0 .. packet.len - first], packet[first..]);
@@ -56,7 +61,42 @@ pub const Ring = struct {
5661 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;
5762 s.count += 1;
5863 s.used += packet.len;
59 return true;
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;
60100 }
61101 /// Explicit owned snapshot copy; data and metadata views otherwise expire
62102 /// on the next mutation. Returns the active destination slice.
@@ -112,3 +152,40 @@ test "initialization rejects zero limits and frees partial allocations" {
112152 };
113153 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
114154}
155
156test "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}
178test "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}