ReviewOS
Reload

also looking at this

home-lang/home

feat(media): add bounded PCM capture, owned snapshots and A/V streaming

#917
Merged chris wants to merge feat/pcm-av-streaming into main
11 files +683 -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/STREAMING_AV.mdmodified+12-0
@@ -44,3 +44,15 @@ zig test packages/video/src/containers/av_stream.zig
4444
4545The Linux native smoke explicitly opens ALSA's `null` test device, never the
4646user's default microphone. See `.github/workflows/streaming-av.yml`.
47
48`Queue.pushRolling` adds bounded circular PCM history. It validates the entire
49input before evicting whole oldest blocks to meet both sample and descriptor
50limits. `retention_floor` reports the end of the discarded prefix. Invalid or
51oversized blocks leave history untouched. `window(minimum, cut, cursor)` reads
52non-destructively with sample-exact clipping, including arena wrap. Each replay
53uses a fresh cursor; chunks and cursors expire on mutation. `Chunk.end_ns` uses
54the original block clock to avoid accumulating fractional sample rounding.
55Consumers can align an encoded keyframe ring to the PCM retention floor and
56replay separate snapshots without consuming live audio. If eviction passes a
57live consumer's watermark, report loss rather than silently producing a gap.
58No encoder, thread or file I/O is performed by these history methods.
packages/video/src/core/pcm_timeline.zigmodified+82-2
@@ -12,7 +12,8 @@ pub fn before(origin: u64, cut: u64, rate: u32, limit: usize) usize {
1212 return @intCast(@min(limit, (@as(u128, cut - origin) * rate + 999_999_999) / 1_000_000_000));
1313}
1414const Entry = struct { offset: usize, frames: usize, consumed: usize = 0, timestamp: u64 };
15pub const Chunk = struct { data: []const u8, timestamp: u64, frames: usize };
15pub const Chunk = struct { data: []const u8, timestamp: u64, frames: usize, end_ns: u64 };
16pub const Cursor = struct { block: usize = 0, frame: usize = 0 };
1617pub const Queue = struct {
1718 allocator: std.mem.Allocator,
1819 data: []u8,
@@ -26,6 +27,7 @@ pub const Queue = struct {
2627 write_pos: usize = 0,
2728 last_end: ?u64 = null,
2829 trimmed: u64 = 0,
30 retention_floor: ?u64 = null,
2931 pub fn init(a: std.mem.Allocator, rate: u32, channels: u32, frames: usize, blocks: usize) !Queue {
3032 if (rate < 8000 or rate > 192000 or channels < 1 or channels > 2 or frames == 0 or frames > rate * 10 or blocks == 0 or blocks > 4096) return error.InvalidConfiguration;
3133 const data = try a.alloc(u8, frames * channels * 2);
@@ -62,6 +64,44 @@ pub const Queue = struct {
6264 s.count -= 1;
6365 }
6466 }
67 /// Circular ownership: reject invalid input before eviction, then discard
68 /// whole oldest blocks until the new block fits. No allocation after init.
69 pub fn pushRolling(s: *Queue, data: []const u8, timestamp: u64) !void {
70 const stride = s.channels * 2;
71 if (data.len == 0 or data.len % stride != 0) return error.InvalidPcm;
72 const frames = data.len / stride;
73 if (frames > s.capacity) return error.BufferFull;
74 if (s.last_end) |end| if (timestamp < end) return error.InvalidTimestamp;
75 _ = try at(timestamp, frames, s.rate);
76 while (frames > s.capacity - s.used or s.count == s.entries.len) {
77 const e = &s.entries[s.head];
78 const n = e.frames - e.consumed;
79 s.retention_floor = try at(e.timestamp, e.frames, s.rate);
80 s.trimmed += n;
81 s.consume(e, n);
82 }
83 try s.push(data, timestamp);
84 }
85 /// Non-consuming sample window. Cursor and returned slices expire on any
86 /// mutation. Separate cursors can replay the same history independently.
87 pub fn window(s: *const Queue, minimum: u64, cut: u64, cursor: *Cursor) !?Chunk {
88 while (cursor.block < s.count) {
89 const e = s.entries[(s.head + cursor.block) % s.entries.len];
90 const begin = @max(@max(e.consumed, cursor.frame), before(e.timestamp, minimum, s.rate, e.frames));
91 const end = before(e.timestamp, cut, s.rate, e.frames);
92 if (begin >= end) {
93 if (end < e.frames) return null;
94 cursor.block += 1;
95 cursor.frame = 0;
96 continue;
97 }
98 const offset = (e.offset + begin) % s.capacity;
99 const n = @min(end - begin, s.capacity - offset);
100 cursor.frame = begin + n;
101 return .{ .data = s.data[offset * s.channels * 2 ..][0 .. n * s.channels * 2], .timestamp = try at(e.timestamp, begin, s.rate), .frames = n, .end_ns = try at(e.timestamp, begin + n, s.rate) };
102 }
103 return null;
104 }
65105 /// Emit samples with time >= minimum and < cut. Wrap is returned as separate
66106 /// contiguous chunks; never duplicate a sample at a segment boundary.
67107 pub fn take(s: *Queue, minimum: u64, cut: u64) !?Chunk {
@@ -78,7 +118,7 @@ pub const Queue = struct {
78118 if (end <= e.consumed) return null;
79119 const offset = (e.offset + e.consumed) % s.capacity;
80120 const n = @min(end - e.consumed, s.capacity - offset);
81 const result = Chunk{ .data = s.data[offset * s.channels * 2 ..][0 .. n * s.channels * 2], .timestamp = try at(e.timestamp, e.consumed, s.rate), .frames = n };
121 const result = Chunk{ .data = s.data[offset * s.channels * 2 ..][0 .. n * s.channels * 2], .timestamp = try at(e.timestamp, e.consumed, s.rate), .frames = n, .end_ns = try at(e.timestamp, e.consumed + n, s.rate) };
82122 s.consume(e, n);
83123 return result;
84124 }
@@ -124,3 +164,43 @@ test "PCM partial allocation failure cleanup" {
124164 };
125165 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
126166}
167
168test "rolling PCM snapshots are repeatable across wrap and exact sample boundaries" {
169 var q = try Queue.init(std.testing.allocator, 44100, 1, 8, 3);
170 defer q.deinit();
171 try q.pushRolling(&.{ 1, 0, 2, 0, 3, 0, 4, 0, 5, 0 }, 1000000);
172 try q.pushRolling(&.{ 6, 0, 7, 0, 8, 0, 9, 0 }, try at(1000000, 5, 44100));
173 try std.testing.expectEqual(@as(u64, 5), q.trimmed);
174 try std.testing.expectEqual(@as(usize, 4), q.used);
175 const start = try at(try at(1000000, 5, 44100), 1, 44100);
176 for (0..2) |_| {
177 var cursor: Cursor = .{};
178 const a = (try q.window(start, std.math.maxInt(u64), &cursor)).?;
179 try std.testing.expectEqualSlices(u8, &.{ 7, 0, 8, 0 }, a.data);
180 const b = (try q.window(start, std.math.maxInt(u64), &cursor)).?;
181 try std.testing.expectEqualSlices(u8, &.{ 9, 0 }, b.data);
182 try std.testing.expect((try q.window(start, std.math.maxInt(u64), &cursor)) == null);
183 try std.testing.expectEqual(@as(usize, 4), q.used);
184 }
185 const floor = q.retention_floor;
186 try std.testing.expectError(error.InvalidTimestamp, q.pushRolling(&.{ 1, 0 }, 0));
187 var oversized: [18]u8 = undefined;
188 @memset(&oversized, 0);
189 try std.testing.expectError(error.BufferFull, q.pushRolling(&oversized, q.last_end.?));
190 try std.testing.expectEqual(floor, q.retention_floor);
191 try std.testing.expectEqual(@as(usize, 4), q.used);
192}
193test "rolling PCM metadata limit and window clipping preserve gaps" {
194 var q = try Queue.init(std.testing.allocator, 8000, 2, 16, 2);
195 defer q.deinit();
196 const block = [_]u8{ 1, 0, 2, 0, 3, 0, 4, 0 };
197 try q.pushRolling(&block, 0);
198 try q.pushRolling(&block, 1000000);
199 try q.pushRolling(&block, 2000000);
200 try std.testing.expectEqual(@as(u64, 2), q.trimmed);
201 var cursor: Cursor = .{};
202 const chunk = (try q.window(1125000, 2000000, &cursor)).?;
203 try std.testing.expectEqual(@as(usize, 1), chunk.frames);
204 try std.testing.expectEqual(@as(u64, 1250000), chunk.end_ns);
205 try std.testing.expect((try q.window(1125000, 2000000, &cursor)) == null);
206}