tiny.sys.pulse

1const std = @import("std");2const alloc_phase = @import("alloc_phase");3const builtin = @import("builtin");4const env = @import("env.zig");5const fs = @import("fs.zig");6const net = @import("net.zig");78pub const supported = builtin.os.tag == .linux;910pub const protocol_version: u32 = 32;11const control_channel: u32 = 0xFFFF_FFFF;12const invalid_index: u32 = 0xFFFF_FFFF;13const cookie_length = 256;14const frame_header_length = 20;15const tagged_u32_bytes = 5;16const tagged_arbitrary_overhead_bytes = 5;17const tagged_string_overhead_bytes = 2;18const property_list_overhead_bytes = 2;19const property_entry_overhead_bytes = 13;20const client_name_key = "application.name";21const media_name_key = "media.name";22const media_name_value = "sdfii";23const auth_payload_bytes = 3 * tagged_u32_bytes + tagged_arbitrary_overhead_bytes + cookie_length;24const drain_payload_bytes = 3 * tagged_u32_bytes;2526const command_error: u32 = 0;27const command_reply: u32 = 2;28const command_create_playback_stream: u32 = 3;29const command_drain_playback_stream: u32 = 4;30const command_auth: u32 = 8;31const command_set_client_name: u32 = 9;32const command_request: u32 = 61;3334pub const sample_format_s16le: u8 = 3;35pub const sample_format_float32le: u8 = 5;3637pub 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};5152pub const Limits = struct {53    inbound_control_payload_bytes: usize = 64 * 1024,54    client_name_bytes: usize = 255,55    channels: u8 = 2,56};5758fn 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}6364fn streamPayloadBytes(channels: u8) usize {65    return 101 + @as(usize, channels) * 5;66}6768pub 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,7677    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};105106const ControlLimits = Limits;107const ControlCapacity = Capacity;108109pub const ControlStorage = struct {110    phase: alloc_phase.capacity.Phase,111    capacity: ControlCapacity,112    bytes: []u8,113114    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};118119    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    };200201    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        else206            try allocator.alloc(u8, capacity.storage_bytes);207        return .{208            .phase = .initialization,209            .capacity = capacity,210            .bytes = bytes,211        };212    }213214    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    }219220    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    }227228    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    }235236    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};244245comptime {246    alloc_phase.capacity.requireAllocatorRejectingOwnerShape(ControlStorage);247}248249pub const Spec = struct {250    format: u8 = sample_format_s16le,251    channels: u8 = 2,252    rate: u32 = 48_000,253254    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};263264pub const TagWriter = struct {265    bytes: []u8,266    index: usize = 0,267268    pub const EncodeError = error{ OutputTooSmall, ValueTooLarge };269270    pub fn init(bytes: []u8) TagWriter {271        return .{ .bytes = bytes };272    }273274    pub fn written(self: *const TagWriter) []const u8 {275        return self.bytes[0..self.index];276    }277278    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    }283284    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    }289290    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    }294295    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    }309310    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    }320321    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    }328329    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    }339340    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    }350351    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    }373374    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    }382383    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};390391pub const TagReader = struct {392    bytes: []const u8,393    index: usize = 0,394395    pub fn u32v(self: *TagReader) Error!u32 {396        try self.expect('L');397        return self.rawU32();398    }399400    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    }410411    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    }420421    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    }430431    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    }438439    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    }444445    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};452453pub const StreamInfo = struct {454    channel: u32,455    sink_input: u32,456    missing: u32,457    minreq: u32,458};459460pub 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,466467    pub const Limits: type = ControlLimits;468    pub const Capacity: type = ControlCapacity;469    pub const Storage: type = ControlStorage;470471    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;480481        var storage = try ControlStorage.init(allocator, limits);482        errdefer storage.deinit(allocator);483484        var cookie = @as([cookie_length]u8, @splat(0));485        loadCookie(&cookie);486487        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();492493        var playback = Playback{494            .stream = stream,495            .spec = spec,496            .storage = storage,497        };498499        try playback.authenticate(&cookie);500        try playback.setClientName(name);501        try playback.createStream();502        playback.storage.activate();503        return playback;504    }505506    pub fn close(self: *Playback, allocator: std.mem.Allocator) void {507        self.stream.close();508        self.storage.deinit(allocator);509        self.* = undefined;510    }511512    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());522523        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    }528529    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    }539540    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;543544        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());554555        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    }567568    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    }581582    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    }593594    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    }611612    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    }622623    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    }633634    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    }654655    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    }669670    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    }682683    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    }691692    fn nextSequence(self: *Playback) u32 {693        const value = self.sequence;694        self.sequence += 1;695        return value;696    }697};698699fn 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}737738pub 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}747748fn 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}761762fn 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}775776test "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    }782783    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();791792    try std.testing.expectEqual(@as(u8, 'L'), written[0]);793    try std.testing.expectEqual(@as(usize, auth_payload_bytes), written.len);794795    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}800801test "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();810811    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}821822fn 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;835836    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;841842    writer = TagWriter.init(&payload);843    try writePlaybackStreamPayload(&writer, 0, .{ .channels = channels }, 1024);844    const encoded_stream_payload_bytes = writer.written().len;845846    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}868869test "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    }875876    var name: [4096]u8 = undefined;877    @memset(&name, 'n');878879    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    }895896    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}917918fn 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}926927test "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    }933934    try std.testing.checkAllAllocationFailures(935        std.testing.allocator,936        checkControlStorageInitFailures,937        .{},938    );939940    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}949950test "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    }956957    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    }976977    const pointer = storage.bytes.ptr;978    const capacity = storage.capacity;979    @memset(storage.bytes, 0xa5);980    phase_allocator.seal();981    storage.activate();982983    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);992993    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);10041005    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}10111012test "pulse playback rejects root limits before allocation or connection effects" {1013    if (comptime !supported) return error.SkipZigTest;10141015    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}10491050fn 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}10571058fn 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}10651066test "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    }10721073    if (comptime !supported) return error.SkipZigTest;10741075    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    }11061107    const oversized = @as([33]u8, @splat(0x5a));1108    try writeTestFrame(peer, control_channel, &oversized);1109    try writeTestReply(peer);11101111    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();11191120    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}11421143test "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}11471148test "live playback reaches the pulse server" {1149    if (comptime !supported) return error.SkipZigTest;1150    if (!available()) return error.SkipZigTest;11511152    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);11621163    try std.testing.expect(playback.info.sink_input != invalid_index);1164    try std.testing.expect(playback.info.missing > 0);11651166    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}

1pub const pulse = @import("pulse.zig");