lib/sys/src/pulse.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const builtin = @import("builtin");
4 const env = @import("env.zig");
5 const fs = @import("fs.zig");
6 const net = @import("net.zig");
7
8 pub const supported = builtin.os.tag == .linux;
9
10 pub const protocol_version: u32 = 32;
11 const control_channel: u32 = 0xFFFF_FFFF;
12 const invalid_index: u32 = 0xFFFF_FFFF;
13 const cookie_length = 256;
14 const frame_header_length = 20;
15 const tagged_u32_bytes = 5;
16 const tagged_arbitrary_overhead_bytes = 5;
17 const tagged_string_overhead_bytes = 2;
18 const property_list_overhead_bytes = 2;
19 const property_entry_overhead_bytes = 13;
20 const client_name_key = "application.name";
21 const media_name_key = "media.name";
22 const media_name_value = "sdfii";
23 const auth_payload_bytes = 3 * tagged_u32_bytes + tagged_arbitrary_overhead_bytes + cookie_length;
24 const drain_payload_bytes = 3 * tagged_u32_bytes;
25
26 const command_error: u32 = 0;
27 const command_reply: u32 = 2;
28 const command_create_playback_stream: u32 = 3;
29 const command_drain_playback_stream: u32 = 4;
30 const command_auth: u32 = 8;
31 const command_set_client_name: u32 = 9;
32 const command_request: u32 = 61;
33
34 pub const sample_format_s16le: u8 = 3;
35 pub const sample_format_float32le: u8 = 5;
36
37 pub const Error = error{
38 UnsupportedPlatform,
39 MissingRuntimeDirectory,
40 ConnectionFailed,
41 AuthenticationFailed,
42 CapacityOverflow,
43 ChannelLimitExceeded,
44 ClientNameLimitExceeded,
45 ProtocolError,
46 ControlPayloadCapacityExceeded,
47 ServerError,
48 StreamRejected,
49 OutOfMemory,
50 };
51
52 pub const Limits = struct {
53 inbound_control_payload_bytes: usize = 64 * 1024,
54 client_name_bytes: usize = 255,
55 channels: u8 = 2,
56 };
57
58 fn clientNamePayloadBytes(name_bytes: usize) error{CapacityOverflow}!usize {
59 const fixed_bytes = 2 * tagged_u32_bytes +
60 property_list_overhead_bytes + client_name_key.len + property_entry_overhead_bytes;
61 return std.math.add(usize, fixed_bytes, name_bytes) catch error.CapacityOverflow;
62 }
63
64 fn streamPayloadBytes(channels: u8) usize {
65 return 101 + @as(usize, channels) * 5;
66 }
67
68 pub const Capacity = struct {
69 inbound_control_payload_bytes: usize,
70 auth_payload_bytes: usize,
71 client_name_payload_bytes: usize,
72 stream_payload_bytes: usize,
73 drain_payload_bytes: usize,
74 outbound_control_payload_bytes: usize,
75 storage_bytes: usize,
76
77 pub fn derive(limits: Limits) error{CapacityOverflow}!Capacity {
78 if (limits.inbound_control_payload_bytes > std.math.maxInt(u32)) {
79 return error.CapacityOverflow;
80 }
81 if (limits.client_name_bytes >= std.math.maxInt(u32)) {
82 return error.CapacityOverflow;
83 }
84 const client_name_payload_bytes = try clientNamePayloadBytes(limits.client_name_bytes);
85 if (client_name_payload_bytes > std.math.maxInt(u32)) return error.CapacityOverflow;
86 const stream_payload_bytes = streamPayloadBytes(limits.channels);
87 const outbound_control_payload_bytes = @max(
88 @max(auth_payload_bytes, client_name_payload_bytes),
89 @max(stream_payload_bytes, drain_payload_bytes),
90 );
91 return .{
92 .inbound_control_payload_bytes = limits.inbound_control_payload_bytes,
93 .auth_payload_bytes = auth_payload_bytes,
94 .client_name_payload_bytes = client_name_payload_bytes,
95 .stream_payload_bytes = stream_payload_bytes,
96 .drain_payload_bytes = drain_payload_bytes,
97 .outbound_control_payload_bytes = outbound_control_payload_bytes,
98 .storage_bytes = @max(
99 limits.inbound_control_payload_bytes,
100 outbound_control_payload_bytes,
101 ),
102 };
103 }
104 };
105
106 const ControlLimits = Limits;
107 const ControlCapacity = Capacity;
108
109 pub const ControlStorage = struct {
110 phase: alloc_phase.capacity.Phase,
111 capacity: ControlCapacity,
112 bytes: []u8,
113
114 pub const Limits: type = ControlLimits;
115 pub const Capacity: type = ControlCapacity;
116 pub const Exhaustion = error{ControlPayloadCapacityExceeded};
117 pub const InitError = std.mem.Allocator.Error || error{CapacityOverflow};
118
119 pub const claim: alloc_phase.capacity.Declaration = .{
120 .source = .{
121 .id = "sys.pulse_control_storage",
122 .kind = .phase_static,
123 .limit_source = .caller,
124 .storage = .{
125 .covered = &.{
126 .{
127 .id = "one_reusable_pulseaudio_control_payload_region",
128 .lifetime = .steady,
129 .detail = "one reusable PulseAudio control payload region",
130 },
131 },
132 .excluded = &.{
133 "socket and kernel buffers, stream handles, and PulseAudio server state",
134 "caller-owned PCM frames and fixed frame-header stack values",
135 "process environment, cookie file handle, and fixed cookie stack bytes",
136 },
137 },
138 .capacity = .{
139 .inputs = &.{
140 alloc_phase.capacity.bindInput(ControlLimits, "inbound_control_payload_bytes", "inbound_control_payload_bytes"),
141 alloc_phase.capacity.bindInput(ControlLimits, "client_name_bytes", "client_name_bytes"),
142 alloc_phase.capacity.bindInput(ControlLimits, "channels", "channels"),
143 },
144 .type_selectors = &.{},
145 .nodes = &.{
146 .{ .input = 0 },
147 .{ .input = 1 },
148 .{ .input = 2 },
149 .{ .scale = .{ .node = 2, .coefficient = .{ .literal = 4 } } },
150 .{ .add = .{ .left = 1, .right = 3 } },
151 .{ .maximum = .{ .left = 0, .right = 4 } },
152 },
153 .assertions = &.{.{
154 .scope = .closure_total,
155 .measure = .retained,
156 .relation = .exact,
157 .expression = 5,
158 }},
159 },
160 .overload = .{
161 .kind = .reject_before_mutation,
162 .detail = "oversize payloads reject before reusable bytes mutate",
163 },
164 .risks = .{
165 .transitive = .{
166 .status = .open,
167 .detail = "playback control traverses separate stream and environment owners",
168 },
169 .foreign = .{
170 .status = .open,
171 .detail = "control I/O enters socket, kernel, and PulseAudio server storage",
172 },
173 },
174 .obligations = &.{
175 .{ .key = "sys_pulse_control_capacity", .role = .capacity_model },
176 .{ .key = "sys_pulse_control_oom_retry", .role = .custom },
177 .{ .key = "sys_pulse_control_sealed", .role = .overload },
178 .{ .key = "sys_pulse_control_semantics", .role = .custom },
179 .{ .key = "sys_pulse_control_drain", .role = .custom },
180 },
181 },
182 .bindings = .{
183 .owner = @This(),
184 .seal = .{
185 .family = alloc_phase.capacity.selector(@This().activate),
186 .premise = .{
187 .class = .checked_semantic_fact,
188 .authority = .checker,
189 },
190 },
191 .teardown = .{
192 .family = alloc_phase.capacity.selector(@This().deinit),
193 .premise = .{
194 .class = .checked_semantic_fact,
195 .authority = .checker,
196 },
197 },
198 },
199 };
200
201 pub fn init(allocator: std.mem.Allocator, limits: ControlLimits) InitError!ControlStorage {
202 const capacity = try ControlCapacity.derive(limits);
203 const bytes = if (capacity.storage_bytes == 0)
204 @as([]u8, &.{})
205 else
206 try allocator.alloc(u8, capacity.storage_bytes);
207 return .{
208 .phase = .initialization,
209 .capacity = capacity,
210 .bytes = bytes,
211 };
212 }
213
214 pub fn activate(self: *ControlStorage) void {
215 std.debug.assert(self.phase == .initialization);
216 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
217 self.phase = .steady;
218 }
219
220 pub fn inbound(self: *ControlStorage, length: usize) Exhaustion![]u8 {
221 std.debug.assert(self.phase != .teardown);
222 if (length > self.capacity.inbound_control_payload_bytes) {
223 return error.ControlPayloadCapacityExceeded;
224 }
225 return self.bytes[0..length];
226 }
227
228 pub fn outbound(self: *ControlStorage, length: usize) Exhaustion![]u8 {
229 std.debug.assert(self.phase != .teardown);
230 if (length > self.capacity.outbound_control_payload_bytes) {
231 return error.ControlPayloadCapacityExceeded;
232 }
233 return self.bytes[0..length];
234 }
235
236 pub fn deinit(self: *ControlStorage, allocator: std.mem.Allocator) void {
237 std.debug.assert(self.phase != .teardown);
238 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
239 self.phase = .teardown;
240 if (self.bytes.len != 0) allocator.free(self.bytes);
241 self.bytes = &.{};
242 }
243 };
244
245 comptime {
246 alloc_phase.capacity.requireAllocatorRejectingOwnerShape(ControlStorage);
247 }
248
249 pub const Spec = struct {
250 format: u8 = sample_format_s16le,
251 channels: u8 = 2,
252 rate: u32 = 48_000,
253
254 pub fn frameBytes(self: Spec) u32 {
255 const sample: u32 = switch (self.format) {
256 sample_format_s16le => 2,
257 sample_format_float32le => 4,
258 else => 2,
259 };
260 return sample * self.channels;
261 }
262 };
263
264 pub const TagWriter = struct {
265 bytes: []u8,
266 index: usize = 0,
267
268 pub const EncodeError = error{ OutputTooSmall, ValueTooLarge };
269
270 pub fn init(bytes: []u8) TagWriter {
271 return .{ .bytes = bytes };
272 }
273
274 pub fn written(self: *const TagWriter) []const u8 {
275 return self.bytes[0..self.index];
276 }
277
278 pub fn u32v(self: *TagWriter, value: u32) EncodeError!void {
279 const target = try self.take(tagged_u32_bytes);
280 target[0] = 'L';
281 std.mem.writeInt(u32, target[1..5], value, .big);
282 }
283
284 pub fn u8v(self: *TagWriter, value: u8) EncodeError!void {
285 const target = try self.take(2);
286 target[0] = 'B';
287 target[1] = value;
288 }
289
290 pub fn boolean(self: *TagWriter, value: bool) EncodeError!void {
291 const target = try self.take(1);
292 target[0] = if (value) '1' else '0';
293 }
294
295 pub fn string(self: *TagWriter, value: ?[]const u8) EncodeError!void {
296 if (value) |text| {
297 const length = std.math.add(usize, text.len, tagged_string_overhead_bytes) catch {
298 return error.ValueTooLarge;
299 };
300 const target = try self.take(length);
301 target[0] = 't';
302 @memcpy(target[1..][0..text.len], text);
303 target[target.len - 1] = 0;
304 } else {
305 const target = try self.take(1);
306 target[0] = 'N';
307 }
308 }
309
310 pub fn arbitrary(self: *TagWriter, bytes: []const u8) EncodeError!void {
311 if (bytes.len > std.math.maxInt(u32)) return error.ValueTooLarge;
312 const length = std.math.add(usize, bytes.len, tagged_arbitrary_overhead_bytes) catch {
313 return error.ValueTooLarge;
314 };
315 const target = try self.take(length);
316 target[0] = 'x';
317 std.mem.writeInt(u32, target[1..5], @intCast(bytes.len), .big);
318 @memcpy(target[5..], bytes);
319 }
320
321 pub fn sampleSpec(self: *TagWriter, spec: Spec) EncodeError!void {
322 const target = try self.take(7);
323 target[0] = 'a';
324 target[1] = spec.format;
325 target[2] = spec.channels;
326 std.mem.writeInt(u32, target[3..7], spec.rate, .big);
327 }
328
329 pub fn channelMap(self: *TagWriter, channels: u8) EncodeError!void {
330 const target = try self.take(@as(usize, channels) + 2);
331 target[0] = 'm';
332 target[1] = channels;
333 var index: u8 = 0;
334 while (index < channels) : (index += 1) {
335 const position: u8 = if (channels == 1) 0 else 1 + index;
336 target[2 + @as(usize, index)] = position;
337 }
338 }
339
340 pub fn cvolume(self: *TagWriter, channels: u8, volume: u32) EncodeError!void {
341 const target = try self.take(2 + @as(usize, channels) * 4);
342 target[0] = 'v';
343 target[1] = channels;
344 var index: u8 = 0;
345 while (index < channels) : (index += 1) {
346 const offset = 2 + @as(usize, index) * 4;
347 std.mem.writeInt(u32, target[offset..][0..4], volume, .big);
348 }
349 }
350
351 pub fn propList(self: *TagWriter, pairs: []const [2][]const u8) EncodeError!void {
352 var length: usize = property_list_overhead_bytes;
353 for (pairs) |pair| {
354 if (pair[1].len >= std.math.maxInt(u32)) return error.ValueTooLarge;
355 length = std.math.add(usize, length, pair[0].len) catch return error.ValueTooLarge;
356 length = std.math.add(usize, length, pair[1].len) catch return error.ValueTooLarge;
357 length = std.math.add(usize, length, property_entry_overhead_bytes) catch {
358 return error.ValueTooLarge;
359 };
360 }
361 const target = try self.take(length);
362 var writer = TagWriter.init(target);
363 (try writer.take(1))[0] = 'P';
364 for (pairs) |pair| {
365 try writer.string(pair[0]);
366 const data_len: u32 = @intCast(pair[1].len + 1);
367 try writer.u32v(data_len);
368 try writer.propertyValue(pair[1]);
369 }
370 try writer.string(null);
371 std.debug.assert(writer.index == target.len);
372 }
373
374 fn propertyValue(self: *TagWriter, value: []const u8) EncodeError!void {
375 if (value.len >= std.math.maxInt(u32)) return error.ValueTooLarge;
376 const target = try self.take(value.len + 6);
377 target[0] = 'x';
378 std.mem.writeInt(u32, target[1..5], @intCast(value.len + 1), .big);
379 @memcpy(target[5..][0..value.len], value);
380 target[target.len - 1] = 0;
381 }
382
383 fn take(self: *TagWriter, length: usize) EncodeError![]u8 {
384 if (length > self.bytes.len - self.index) return error.OutputTooSmall;
385 const target = self.bytes[self.index..][0..length];
386 self.index += length;
387 return target;
388 }
389 };
390
391 pub const TagReader = struct {
392 bytes: []const u8,
393 index: usize = 0,
394
395 pub fn u32v(self: *TagReader) Error!u32 {
396 try self.expect('L');
397 return self.rawU32();
398 }
399
400 pub fn boolean(self: *TagReader) Error!bool {
401 if (self.index >= self.bytes.len) return error.ProtocolError;
402 const tag = self.bytes[self.index];
403 self.index += 1;
404 return switch (tag) {
405 '1' => true,
406 '0' => false,
407 else => error.ProtocolError,
408 };
409 }
410
411 pub fn skipString(self: *TagReader) Error!void {
412 if (self.index >= self.bytes.len) return error.ProtocolError;
413 const tag = self.bytes[self.index];
414 self.index += 1;
415 if (tag == 'N') return;
416 if (tag != 't') return error.ProtocolError;
417 const terminator = std.mem.indexOfScalarPos(u8, self.bytes, self.index, 0) orelse return error.ProtocolError;
418 self.index = terminator + 1;
419 }
420
421 pub fn sampleSpec(self: *TagReader) Error!Spec {
422 try self.expect('a');
423 if (self.index + 6 > self.bytes.len) return error.ProtocolError;
424 const format = self.bytes[self.index];
425 const channels = self.bytes[self.index + 1];
426 self.index += 2;
427 const rate = try self.rawU32();
428 return .{ .format = format, .channels = channels, .rate = rate };
429 }
430
431 pub fn skipChannelMap(self: *TagReader) Error!void {
432 try self.expect('m');
433 if (self.index >= self.bytes.len) return error.ProtocolError;
434 const channels = self.bytes[self.index];
435 self.index += 1 + @as(usize, channels);
436 if (self.index > self.bytes.len) return error.ProtocolError;
437 }
438
439 fn expect(self: *TagReader, tag: u8) Error!void {
440 if (self.index >= self.bytes.len) return error.ProtocolError;
441 if (self.bytes[self.index] != tag) return error.ProtocolError;
442 self.index += 1;
443 }
444
445 fn rawU32(self: *TagReader) Error!u32 {
446 if (self.index + 4 > self.bytes.len) return error.ProtocolError;
447 const value = std.mem.readInt(u32, self.bytes[self.index..][0..4], .big);
448 self.index += 4;
449 return value;
450 }
451 };
452
453 pub const StreamInfo = struct {
454 channel: u32,
455 sink_input: u32,
456 missing: u32,
457 minreq: u32,
458 };
459
460 pub const Playback = struct {
461 stream: net.Stream,
462 sequence: u32 = 0,
463 info: StreamInfo = .{ .channel = 0, .sink_input = invalid_index, .missing = 0, .minreq = 0 },
464 spec: Spec,
465 storage: ControlStorage,
466
467 pub const Limits: type = ControlLimits;
468 pub const Capacity: type = ControlCapacity;
469 pub const Storage: type = ControlStorage;
470
471 pub fn open(
472 allocator: std.mem.Allocator,
473 limits: ControlLimits,
474 spec: Spec,
475 name: []const u8,
476 ) Error!Playback {
477 if (comptime !supported) return error.UnsupportedPlatform;
478 if (name.len > limits.client_name_bytes) return error.ClientNameLimitExceeded;
479 if (spec.channels > limits.channels) return error.ChannelLimitExceeded;
480
481 var storage = try ControlStorage.init(allocator, limits);
482 errdefer storage.deinit(allocator);
483
484 var cookie = @as([cookie_length]u8, @splat(0));
485 loadCookie(&cookie);
486
487 var path_buffer: [256]u8 = undefined;
488 const path = socketPath(&path_buffer) orelse return error.MissingRuntimeDirectory;
489 const address = net.Address.initUnix(path) catch return error.ConnectionFailed;
490 var stream = net.connectStream(address) catch return error.ConnectionFailed;
491 errdefer stream.close();
492
493 var playback = Playback{
494 .stream = stream,
495 .spec = spec,
496 .storage = storage,
497 };
498
499 try playback.authenticate(&cookie);
500 try playback.setClientName(name);
501 try playback.createStream();
502 playback.storage.activate();
503 return playback;
504 }
505
506 pub fn close(self: *Playback, allocator: std.mem.Allocator) void {
507 self.stream.close();
508 self.storage.deinit(allocator);
509 self.* = undefined;
510 }
511
512 fn authenticate(self: *Playback, cookie: *const [cookie_length]u8) Error!void {
513 var writer = TagWriter.init(
514 self.storage.outbound(self.storage.capacity.auth_payload_bytes) catch unreachable,
515 );
516 writer.u32v(command_auth) catch unreachable;
517 writer.u32v(self.nextSequence()) catch unreachable;
518 writer.u32v(protocol_version) catch unreachable;
519 writer.arbitrary(cookie) catch unreachable;
520 std.debug.assert(writer.written().len == self.storage.capacity.auth_payload_bytes);
521 try self.sendControl(writer.written());
522
523 const reply = try self.readReply();
524 var reader = TagReader{ .bytes = reply };
525 const server_version = try reader.u32v();
526 if ((server_version & 0xFFFF) < 13) return error.AuthenticationFailed;
527 }
528
529 fn setClientName(self: *Playback, name: []const u8) Error!void {
530 const payload_bytes = clientNamePayloadBytes(name.len) catch unreachable;
531 var writer = TagWriter.init(self.storage.outbound(payload_bytes) catch unreachable);
532 writer.u32v(command_set_client_name) catch unreachable;
533 writer.u32v(self.nextSequence()) catch unreachable;
534 writer.propList(&.{.{ client_name_key, name }}) catch unreachable;
535 std.debug.assert(writer.written().len == payload_bytes);
536 try self.sendControl(writer.written());
537 _ = try self.readReply();
538 }
539
540 fn createStream(self: *Playback) Error!void {
541 const bytes_per_second = self.spec.rate * self.spec.frameBytes();
542 const target_length = bytes_per_second / 4;
543
544 const payload_bytes = streamPayloadBytes(self.spec.channels);
545 var writer = TagWriter.init(self.storage.outbound(payload_bytes) catch unreachable);
546 writePlaybackStreamPayload(
547 &writer,
548 self.nextSequence(),
549 self.spec,
550 target_length,
551 ) catch unreachable;
552 std.debug.assert(writer.written().len == payload_bytes);
553 try self.sendControl(writer.written());
554
555 const reply = try self.readReply();
556 var reader = TagReader{ .bytes = reply };
557 self.info.channel = try reader.u32v();
558 self.info.sink_input = try reader.u32v();
559 self.info.missing = try reader.u32v();
560 _ = try reader.u32v();
561 const target_reply = try reader.u32v();
562 _ = try reader.u32v();
563 self.info.minreq = try reader.u32v();
564 _ = target_reply;
565 if (self.info.sink_input == invalid_index) return error.StreamRejected;
566 }
567
568 pub fn writeFrames(self: *Playback, bytes: []const u8) Error!void {
569 var remaining = bytes;
570 while (remaining.len > 0) {
571 if (self.info.missing == 0) {
572 try self.pumpUntilRequest();
573 continue;
574 }
575 const chunk_len = @min(remaining.len, self.info.missing);
576 try self.sendData(remaining[0..chunk_len]);
577 self.info.missing -= @intCast(chunk_len);
578 remaining = remaining[chunk_len..];
579 }
580 }
581
582 pub fn drain(self: *Playback) Error!void {
583 var writer = TagWriter.init(
584 self.storage.outbound(self.storage.capacity.drain_payload_bytes) catch unreachable,
585 );
586 writer.u32v(command_drain_playback_stream) catch unreachable;
587 writer.u32v(self.nextSequence()) catch unreachable;
588 writer.u32v(self.info.channel) catch unreachable;
589 std.debug.assert(writer.written().len == self.storage.capacity.drain_payload_bytes);
590 try self.sendControl(writer.written());
591 _ = try self.readReply();
592 }
593
594 fn pumpUntilRequest(self: *Playback) Error!void {
595 const message = try self.readControl();
596 var reader = TagReader{ .bytes = message };
597 const command = try reader.u32v();
598 _ = try reader.u32v();
599 switch (command) {
600 command_request => {
601 const channel = try reader.u32v();
602 const nbytes = try reader.u32v();
603 if (channel == self.info.channel) {
604 self.info.missing += nbytes;
605 }
606 },
607 command_error => return error.ServerError,
608 else => {},
609 }
610 }
611
612 fn sendData(self: *Playback, bytes: []const u8) Error!void {
613 var header: [frame_header_length]u8 = undefined;
614 std.mem.writeInt(u32, header[0..4], @intCast(bytes.len), .big);
615 std.mem.writeInt(u32, header[4..8], self.info.channel, .big);
616 std.mem.writeInt(u32, header[8..12], 0, .big);
617 std.mem.writeInt(u32, header[12..16], 0, .big);
618 std.mem.writeInt(u32, header[16..20], 0, .big);
619 self.stream.writeAll(&header) catch return error.ConnectionFailed;
620 self.stream.writeAll(bytes) catch return error.ConnectionFailed;
621 }
622
623 fn sendControl(self: *Playback, payload: []const u8) Error!void {
624 var header: [frame_header_length]u8 = undefined;
625 std.mem.writeInt(u32, header[0..4], @intCast(payload.len), .big);
626 std.mem.writeInt(u32, header[4..8], control_channel, .big);
627 std.mem.writeInt(u32, header[8..12], 0, .big);
628 std.mem.writeInt(u32, header[12..16], 0, .big);
629 std.mem.writeInt(u32, header[16..20], 0, .big);
630 self.stream.writeAll(&header) catch return error.ConnectionFailed;
631 self.stream.writeAll(payload) catch return error.ConnectionFailed;
632 }
633
634 fn readReply(self: *Playback) Error![]const u8 {
635 while (true) {
636 const message = try self.readControl();
637 var reader = TagReader{ .bytes = message };
638 const command = try reader.u32v();
639 _ = try reader.u32v();
640 switch (command) {
641 command_reply => return message[reader.index..],
642 command_error => return error.ServerError,
643 command_request => {
644 const channel = try reader.u32v();
645 const nbytes = try reader.u32v();
646 if (channel == self.info.channel) {
647 self.info.missing += nbytes;
648 }
649 },
650 else => {},
651 }
652 }
653 }
654
655 fn readControl(self: *Playback) Error![]const u8 {
656 while (true) {
657 var header: [frame_header_length]u8 = undefined;
658 try self.readExact(&header);
659 const length = std.mem.readInt(u32, header[0..4], .big);
660 const channel = std.mem.readInt(u32, header[4..8], .big);
661 const target = self.storage.inbound(length) catch |err| {
662 try self.discardExact(length, &header);
663 return err;
664 };
665 try self.readExact(target);
666 if (channel == control_channel) return target;
667 }
668 }
669
670 fn discardExact(
671 self: *Playback,
672 length: usize,
673 scratch: *[frame_header_length]u8,
674 ) Error!void {
675 var remaining = length;
676 while (remaining > 0) {
677 const chunk_bytes = @min(remaining, scratch.len);
678 try self.readExact(scratch[0..chunk_bytes]);
679 remaining -= chunk_bytes;
680 }
681 }
682
683 fn readExact(self: *Playback, buffer: []u8) Error!void {
684 var filled: usize = 0;
685 while (filled < buffer.len) {
686 const count = self.stream.read(buffer[filled..]) catch return error.ConnectionFailed;
687 if (count == 0) return error.ConnectionFailed;
688 filled += count;
689 }
690 }
691
692 fn nextSequence(self: *Playback) u32 {
693 const value = self.sequence;
694 self.sequence += 1;
695 return value;
696 }
697 };
698
699 fn writePlaybackStreamPayload(
700 writer: *TagWriter,
701 sequence: u32,
702 spec: Spec,
703 target: u32,
704 ) TagWriter.EncodeError!void {
705 try writer.u32v(command_create_playback_stream);
706 try writer.u32v(sequence);
707 try writer.sampleSpec(spec);
708 try writer.channelMap(spec.channels);
709 try writer.u32v(invalid_index);
710 try writer.string(null);
711 try writer.u32v(target * 4);
712 try writer.boolean(false);
713 try writer.u32v(target);
714 try writer.u32v(target / 2);
715 try writer.u32v(spec.frameBytes() * 256);
716 try writer.u32v(0);
717 try writer.cvolume(spec.channels, 0x10000);
718 try writer.boolean(false);
719 try writer.boolean(false);
720 try writer.boolean(false);
721 try writer.boolean(false);
722 try writer.boolean(false);
723 try writer.boolean(false);
724 try writer.boolean(false);
725 try writer.boolean(false);
726 try writer.boolean(true);
727 try writer.propList(&.{.{ media_name_key, media_name_value }});
728 try writer.boolean(false);
729 try writer.boolean(false);
730 try writer.boolean(false);
731 try writer.boolean(false);
732 try writer.boolean(false);
733 try writer.boolean(false);
734 try writer.boolean(false);
735 try writer.u8v(0);
736 }
737
738 pub fn available() bool {
739 if (comptime !supported) return false;
740 var path_buffer: [256]u8 = undefined;
741 const path = socketPath(&path_buffer) orelse return false;
742 const address = net.Address.initUnix(path) catch return false;
743 var probe = net.connectStream(address) catch return false;
744 probe.close();
745 return true;
746 }
747
748 fn socketPath(buffer: []u8) ?[]const u8 {
749 if (env.get("PULSE_SERVER")) |server| {
750 if (std.mem.startsWith(u8, server, "unix:")) {
751 const path = server[5..];
752 if (path.len == 0 or path.len > buffer.len) return null;
753 @memcpy(buffer[0..path.len], path);
754 return buffer[0..path.len];
755 }
756 return null;
757 }
758 const runtime = env.get("XDG_RUNTIME_DIR") orelse return null;
759 return std.fmt.bufPrint(buffer, "{s}/pulse/native", .{runtime}) catch null;
760 }
761
762 fn loadCookie(cookie: *[cookie_length]u8) void {
763 var path_buffer: [512]u8 = undefined;
764 const home = env.get("HOME") orelse return;
765 const path = std.fmt.bufPrint(&path_buffer, "{s}/.config/pulse/cookie", .{home}) catch return;
766 const file = fs.openAbsoluteFile(path, .{}) catch return;
767 defer fs.closeHandle(file);
768 var filled: usize = 0;
769 while (filled < cookie.len) {
770 const count = fs.readHandle(file, cookie[filled..]) catch return;
771 if (count == 0) return;
772 filled += count;
773 }
774 }
775
776 test "pulse fixed tag writer preserves control encoding semantics" {
777 comptime {
778 alloc_phase.capacity.record(
779 alloc_phase.capacity.witness(ControlStorage, "sys_pulse_control_semantics"),
780 );
781 }
782
783 var payload: [auth_payload_bytes]u8 = undefined;
784 var writer = TagWriter.init(&payload);
785 try writer.u32v(command_auth);
786 try writer.u32v(1);
787 try writer.u32v(protocol_version);
788 const cookie = @as([cookie_length]u8, @splat(0xAB));
789 try writer.arbitrary(&cookie);
790 const written = writer.written();
791
792 try std.testing.expectEqual(@as(u8, 'L'), written[0]);
793 try std.testing.expectEqual(@as(usize, auth_payload_bytes), written.len);
794
795 var reader = TagReader{ .bytes = written };
796 try std.testing.expectEqual(command_auth, try reader.u32v());
797 try std.testing.expectEqual(@as(u32, 1), try reader.u32v());
798 try std.testing.expectEqual(protocol_version, try reader.u32v());
799 }
800
801 test "sample spec and strings round-trip" {
802 var payload: [64]u8 = undefined;
803 var writer = TagWriter.init(&payload);
804 const spec = Spec{ .format = sample_format_s16le, .channels = 2, .rate = 48_000 };
805 try writer.sampleSpec(spec);
806 try writer.string("sdfii");
807 try writer.string(null);
808 try writer.channelMap(2);
809 const written = writer.written();
810
811 var reader = TagReader{ .bytes = written };
812 const decoded = try reader.sampleSpec();
813 try std.testing.expectEqual(spec.format, decoded.format);
814 try std.testing.expectEqual(spec.channels, decoded.channels);
815 try std.testing.expectEqual(spec.rate, decoded.rate);
816 try reader.skipString();
817 try reader.skipString();
818 try reader.skipChannelMap();
819 try std.testing.expectEqual(written.len, reader.index);
820 }
821
822 fn independentlyEncodedCapacity(
823 name: []const u8,
824 inbound_control_payload_bytes: usize,
825 channels: u8,
826 ) !Capacity {
827 var payload: [8192]u8 = undefined;
828 var writer = TagWriter.init(&payload);
829 try writer.u32v(command_auth);
830 try writer.u32v(0);
831 try writer.u32v(protocol_version);
832 const cookie = @as([cookie_length]u8, @splat(0));
833 try writer.arbitrary(&cookie);
834 const encoded_auth_payload_bytes = writer.written().len;
835
836 writer = TagWriter.init(&payload);
837 try writer.u32v(command_set_client_name);
838 try writer.u32v(0);
839 try writer.propList(&.{.{ client_name_key, name }});
840 const encoded_client_name_payload_bytes = writer.written().len;
841
842 writer = TagWriter.init(&payload);
843 try writePlaybackStreamPayload(&writer, 0, .{ .channels = channels }, 1024);
844 const encoded_stream_payload_bytes = writer.written().len;
845
846 writer = TagWriter.init(&payload);
847 try writer.u32v(command_drain_playback_stream);
848 try writer.u32v(0);
849 try writer.u32v(0);
850 const encoded_drain_payload_bytes = writer.written().len;
851 const outbound_control_payload_bytes = @max(
852 @max(encoded_auth_payload_bytes, encoded_client_name_payload_bytes),
853 @max(encoded_stream_payload_bytes, encoded_drain_payload_bytes),
854 );
855 return .{
856 .inbound_control_payload_bytes = inbound_control_payload_bytes,
857 .auth_payload_bytes = encoded_auth_payload_bytes,
858 .client_name_payload_bytes = encoded_client_name_payload_bytes,
859 .stream_payload_bytes = encoded_stream_payload_bytes,
860 .drain_payload_bytes = encoded_drain_payload_bytes,
861 .outbound_control_payload_bytes = outbound_control_payload_bytes,
862 .storage_bytes = @max(
863 inbound_control_payload_bytes,
864 outbound_control_payload_bytes,
865 ),
866 };
867 }
868
869 test "pulse control capacity matches independently encoded payload maxima" {
870 comptime {
871 alloc_phase.capacity.record(
872 alloc_phase.capacity.witness(ControlStorage, "sys_pulse_control_capacity"),
873 );
874 }
875
876 var name: [4096]u8 = undefined;
877 @memset(&name, 'n');
878
879 for (0..name.len + 1) |name_bytes| {
880 const channels: u8 = @intCast(name_bytes % 256);
881 const inbound_control_payload_bytes = name_bytes / 2;
882 try std.testing.expectEqual(
883 try independentlyEncodedCapacity(
884 name[0..name_bytes],
885 inbound_control_payload_bytes,
886 channels,
887 ),
888 try Capacity.derive(.{
889 .inbound_control_payload_bytes = inbound_control_payload_bytes,
890 .client_name_bytes = name_bytes,
891 .channels = channels,
892 }),
893 );
894 }
895
896 const client_name_base_bytes = try clientNamePayloadBytes(0);
897 const maximum_client_name_bytes = @as(usize, std.math.maxInt(u32)) -
898 client_name_base_bytes;
899 _ = try Capacity.derive(.{
900 .inbound_control_payload_bytes = std.math.maxInt(u32),
901 .client_name_bytes = maximum_client_name_bytes,
902 .channels = std.math.maxInt(u8),
903 });
904 try std.testing.expectError(
905 error.CapacityOverflow,
906 Capacity.derive(.{ .client_name_bytes = maximum_client_name_bytes + 1 }),
907 );
908 if (@sizeOf(usize) > @sizeOf(u32)) {
909 try std.testing.expectError(
910 error.CapacityOverflow,
911 Capacity.derive(.{
912 .inbound_control_payload_bytes = @as(usize, std.math.maxInt(u32)) + 1,
913 }),
914 );
915 }
916 }
917
918 fn checkControlStorageInitFailures(allocator: std.mem.Allocator) !void {
919 var storage = try ControlStorage.init(allocator, .{
920 .inbound_control_payload_bytes = 4096,
921 .client_name_bytes = 32,
922 .channels = 8,
923 });
924 storage.deinit(allocator);
925 }
926
927 test "pulse control storage retries after allocation failure" {
928 comptime {
929 alloc_phase.capacity.record(
930 alloc_phase.capacity.witness(ControlStorage, "sys_pulse_control_oom_retry"),
931 );
932 }
933
934 try std.testing.checkAllAllocationFailures(
935 std.testing.allocator,
936 checkControlStorageInitFailures,
937 .{},
938 );
939
940 var storage = try ControlStorage.init(std.testing.allocator, .{
941 .inbound_control_payload_bytes = 4096,
942 .client_name_bytes = 32,
943 .channels = 8,
944 });
945 defer storage.deinit(std.testing.allocator);
946 storage.activate();
947 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, storage.phase);
948 }
949
950 test "pulse control storage is sealed before encoding and boundary rejection" {
951 comptime {
952 alloc_phase.capacity.record(
953 alloc_phase.capacity.witness(ControlStorage, "sys_pulse_control_sealed"),
954 );
955 }
956
957 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
958 var storage = ControlStorage.init(
959 phase_allocator.initializationAllocator(),
960 .{
961 .inbound_control_payload_bytes = 32,
962 .client_name_bytes = 8,
963 .channels = 2,
964 },
965 ) catch |err| {
966 phase_allocator.abortInitialization();
967 phase_allocator.deinit();
968 return err;
969 };
970 defer {
971 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
972 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
973 if (storage.phase != .teardown) storage.deinit(phase_allocator.teardownAllocator());
974 phase_allocator.deinit();
975 }
976
977 const pointer = storage.bytes.ptr;
978 const capacity = storage.capacity;
979 @memset(storage.bytes, 0xa5);
980 phase_allocator.seal();
981 storage.activate();
982
983 try std.testing.expectError(
984 error.ControlPayloadCapacityExceeded,
985 storage.inbound(storage.capacity.inbound_control_payload_bytes + 1),
986 );
987 try std.testing.expectError(
988 error.ControlPayloadCapacityExceeded,
989 storage.outbound(storage.capacity.outbound_control_payload_bytes + 1),
990 );
991 for (storage.bytes) |byte| try std.testing.expectEqual(@as(u8, 0xa5), byte);
992
993 var writer = TagWriter.init(
994 try storage.outbound(storage.capacity.auth_payload_bytes),
995 );
996 try writer.u32v(command_auth);
997 try writer.u32v(0);
998 try writer.u32v(protocol_version);
999 const cookie = @as([cookie_length]u8, @splat(0));
1000 try writer.arbitrary(&cookie);
1001 try std.testing.expectEqual(storage.capacity.auth_payload_bytes, writer.written().len);
1002 try std.testing.expect(storage.bytes.ptr == pointer);
1003 try std.testing.expectEqual(capacity, storage.capacity);
1004
1005 var short = @as([(tagged_u32_bytes - 1)]u8, @splat(0xa5));
1006 var short_writer = TagWriter.init(&short);
1007 try std.testing.expectError(error.OutputTooSmall, short_writer.u32v(1));
1008 try std.testing.expectEqual(@as(usize, 0), short_writer.written().len);
1009 for (short) |byte| try std.testing.expectEqual(@as(u8, 0xa5), byte);
1010 }
1011
1012 test "pulse playback rejects root limits before allocation or connection effects" {
1013 if (comptime !supported) return error.SkipZigTest;
1014
1015 var failing = std.testing.FailingAllocator.init(
1016 std.testing.allocator,
1017 .{ .fail_index = 0 },
1018 );
1019 try std.testing.expectError(
1020 error.ClientNameLimitExceeded,
1021 Playback.open(
1022 failing.allocator(),
1023 .{ .client_name_bytes = 3 },
1024 .{},
1025 "four",
1026 ),
1027 );
1028 try std.testing.expectError(
1029 error.ChannelLimitExceeded,
1030 Playback.open(
1031 failing.allocator(),
1032 .{ .channels = 1 },
1033 .{ .channels = 2 },
1034 "",
1035 ),
1036 );
1037 const maximum_client_name_bytes = @as(usize, std.math.maxInt(u32)) -
1038 (try clientNamePayloadBytes(0));
1039 try std.testing.expectError(
1040 error.CapacityOverflow,
1041 Playback.open(
1042 failing.allocator(),
1043 .{ .client_name_bytes = maximum_client_name_bytes + 1 },
1044 .{},
1045 "",
1046 ),
1047 );
1048 }
1049
1050 fn writeTestFrame(peer: net.Stream, channel: u32, payload: []const u8) !void {
1051 var header = @as([frame_header_length]u8, @splat(0));
1052 std.mem.writeInt(u32, header[0..4], @intCast(payload.len), .big);
1053 std.mem.writeInt(u32, header[4..8], channel, .big);
1054 try peer.writeAll(&header);
1055 try peer.writeAll(payload);
1056 }
1057
1058 fn writeTestReply(peer: net.Stream) !void {
1059 var payload: [2 * tagged_u32_bytes]u8 = undefined;
1060 var writer = TagWriter.init(&payload);
1061 try writer.u32v(command_reply);
1062 try writer.u32v(0);
1063 try writeTestFrame(peer, control_channel, &payload);
1064 }
1065
1066 test "pulse playback drains through sealed control storage" {
1067 comptime {
1068 alloc_phase.capacity.record(
1069 alloc_phase.capacity.witness(ControlStorage, "sys_pulse_control_drain"),
1070 );
1071 }
1072
1073 if (comptime !supported) return error.SkipZigTest;
1074
1075 const sockets = try net.socketPairUnixStream();
1076 var peer = net.Stream.initFd(sockets[1]);
1077 defer peer.close();
1078 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
1079 const storage = ControlStorage.init(
1080 phase_allocator.initializationAllocator(),
1081 .{
1082 .inbound_control_payload_bytes = 32,
1083 .client_name_bytes = 8,
1084 .channels = 2,
1085 },
1086 ) catch |err| {
1087 net.close(sockets[0]);
1088 phase_allocator.abortInitialization();
1089 phase_allocator.deinit();
1090 return err;
1091 };
1092 var playback = Playback{
1093 .stream = net.Stream.initFd(sockets[0]),
1094 .info = .{ .channel = 7, .sink_input = 0, .missing = 0, .minreq = 0 },
1095 .spec = .{},
1096 .storage = storage,
1097 };
1098 defer {
1099 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
1100 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
1101 if (playback.storage.phase != .teardown) {
1102 playback.close(phase_allocator.teardownAllocator());
1103 }
1104 phase_allocator.deinit();
1105 }
1106
1107 const oversized = @as([33]u8, @splat(0x5a));
1108 try writeTestFrame(peer, control_channel, &oversized);
1109 try writeTestReply(peer);
1110
1111 const pointer = playback.storage.bytes.ptr;
1112 const capacity = playback.storage.capacity;
1113 @memset(playback.storage.bytes, 0xa5);
1114 phase_allocator.seal();
1115 playback.storage.activate();
1116 try std.testing.expectError(error.ControlPayloadCapacityExceeded, playback.readControl());
1117 for (playback.storage.bytes) |byte| try std.testing.expectEqual(@as(u8, 0xa5), byte);
1118 try playback.drain();
1119
1120 var request: [frame_header_length + drain_payload_bytes]u8 = undefined;
1121 var received: usize = 0;
1122 while (received < request.len) {
1123 const count = try peer.read(request[received..]);
1124 if (count == 0) return error.ConnectionFailed;
1125 received += count;
1126 }
1127 try std.testing.expectEqual(
1128 @as(u32, drain_payload_bytes),
1129 std.mem.readInt(u32, request[0..4], .big),
1130 );
1131 try std.testing.expectEqual(
1132 control_channel,
1133 std.mem.readInt(u32, request[4..8], .big),
1134 );
1135 var reader = TagReader{ .bytes = request[frame_header_length..] };
1136 try std.testing.expectEqual(command_drain_playback_stream, try reader.u32v());
1137 try std.testing.expectEqual(@as(u32, 0), try reader.u32v());
1138 try std.testing.expectEqual(@as(u32, 7), try reader.u32v());
1139 try std.testing.expect(playback.storage.bytes.ptr == pointer);
1140 try std.testing.expectEqual(capacity, playback.storage.capacity);
1141 }
1142
1143 test "spec frame bytes follow format and channels" {
1144 try std.testing.expectEqual(@as(u32, 4), (Spec{ .format = sample_format_s16le, .channels = 2 }).frameBytes());
1145 try std.testing.expectEqual(@as(u32, 8), (Spec{ .format = sample_format_float32le, .channels = 2 }).frameBytes());
1146 }
1147
1148 test "live playback reaches the pulse server" {
1149 if (comptime !supported) return error.SkipZigTest;
1150 if (!available()) return error.SkipZigTest;
1151
1152 const allocator = std.testing.allocator;
1153 const spec = Spec{ .format = sample_format_s16le, .channels = 2, .rate = 48_000 };
1154 var playback = Playback.open(allocator, .{
1155 .client_name_bytes = "sys-pulse-live-test".len,
1156 .channels = spec.channels,
1157 }, spec, "sys-pulse-live-test") catch |err| switch (err) {
1158 error.MissingRuntimeDirectory, error.ConnectionFailed => return error.SkipZigTest,
1159 else => return err,
1160 };
1161 defer playback.close(allocator);
1162
1163 try std.testing.expect(playback.info.sink_input != invalid_index);
1164 try std.testing.expect(playback.info.missing > 0);
1165
1166 const duration_ms: u32 = blk: {
1167 const raw = env.get("TINY_PULSE_LIVE_MS") orelse break :blk 100;
1168 break :blk std.fmt.parseInt(u32, raw, 10) catch 100;
1169 };
1170 const frame_count = spec.rate * duration_ms / 1000;
1171 const bytes = try allocator.alloc(u8, frame_count * spec.frameBytes());
1172 defer allocator.free(bytes);
1173 var index: usize = 0;
1174 while (index < frame_count) : (index += 1) {
1175 const phase = @as(f32, @floatFromInt(index)) / 48_000.0;
1176 const sample: i16 = @intFromFloat(@sin(phase * 2.0 * std.math.pi * 440.0) * 8000.0);
1177 std.mem.writeInt(i16, bytes[index * 4 ..][0..2], sample, .little);
1178 std.mem.writeInt(i16, bytes[index * 4 + 2 ..][0..2], sample, .little);
1179 }
1180 try playback.writeFrames(bytes);
1181 try playback.drain();
1182 }