zzstoatzz.io / websocket.zig
websocket
0
Fork 1

Configure Feed

Select the types of activity you want to include in your feed.

add write lock to websocket client (matches server-side Conn.lock)

client.writeFrame() is called concurrently from:
- pingLoop task (subscriber keepalive)
- readLoop auto-pong (server ping response)
- close path (connection teardown)

without serialization, concurrent writes interleave frame headers/payloads,
corrupting the stream and causing GPF in memcpy → Writer.zig → writeAll.

adds _write_lock: Io.Mutex to Client struct, acquired around the two
writeAll calls in writeFrame(). header construction and payload masking
happen outside the lock to minimize hold time.

the server-side Conn already has this pattern (lock: Io.Mutex around
writeFrame/writeFramed/writeAllIOVec). this brings the client to parity.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

author
zzstoatzz co-author
Claude Opus 4.6 date (Apr 2, 2026, 7:58 PM -0500) commit 0261b7d8 parent edeca263
+12 -25
+12 -25
src/client/client.zig
··· 48 48 _compression_opts: ?CompressionOpts, 49 49 _compression: ?Client.Compression = null, 50 50 51 + // Serializes writes from concurrent tasks (ping loop, auto-pong, close). 52 + // Matches server-side Conn.lock pattern. 53 + _write_lock: Io.Mutex = .init, 54 + 51 55 // When creating a client, we can either be given a BufferProvider or create 52 56 // one ourselves. If we create it ourselves (in init), we "own" it and must 53 57 // free it on deinit. (The reference to the buffer provider is already in the ··· 392 396 pub fn writeFrame(self: *Client, op_code: proto.OpCode, data: []u8) !void { 393 397 const payload = data; 394 398 const compressed = false; 395 - // if (self._compression) |c| { 396 - // if (data.len >= c.write_treshold and (op_code == .binary or op_code == .text)) { 397 - // compressed = true; 398 - 399 - // var writer = &c.writer; 400 - // var compressor = &c.compressor; 401 - // var fbs = std.io.fixedBufferStream(data); 402 - // _ = try compressor.compress(fbs.reader()); 403 - // try compressor.flush(); 404 - // payload = writer.items[0 .. writer.items.len - 4]; 405 - 406 - // if (c.reset) { 407 - // c.compressor = try Compression.Type.init(writer.writer(), .{}); 408 - // } 409 - // } 410 - // } 411 - // defer if (compressed) { 412 - // const c = self._compression.?; 413 - // if (c.retain_writer) { 414 - // c.compressor.wrt.context.clearRetainingCapacity(); 415 - // } else { 416 - // c.compressor.wrt.context.clearAndFree(); 417 - // } 418 - // }; 419 399 420 400 // maximum possible prefix length. op_code + length_type + 8byte length + 4 byte mask 421 401 var buf: [14]u8 = undefined; ··· 428 408 429 409 const mask = self._mask_fn(self.io); 430 410 @memcpy(buf[header_len..header_end], &mask); 431 - try self.stream.writeAll(buf[0..header_end]); 432 411 433 412 if (payload.len > 0) { 434 413 proto.mask(&mask, payload); 414 + } 415 + 416 + // Serialize writes — concurrent ping/pong/close must not interleave frames. 417 + self._write_lock.lockUncancelable(self.io); 418 + defer self._write_lock.unlock(self.io); 419 + 420 + try self.stream.writeAll(buf[0..header_end]); 421 + if (payload.len > 0) { 435 422 try self.stream.writeAll(payload); 436 423 } 437 424 }