StreamablePipe corrupts concurrent large writes and can stall the log watcher

## Summary

`StreamablePipe.write()` calls `multiprocessing.connection.Connection.send_bytes()` without serializing access. A Dramatiq child redirects both `sys.stdout` and `sys.stderr` to one `StreamablePipe`, so independent concurrent writers can interleave pipe frames. This corrupts the parent log watcher stream.

## Impact

For payloads larger than 16 KiB, CPython emits the frame header and payload in separate writes. Interleaving can make `watch_logs()` decode a mixed payload or treat arbitrary bytes as a payload length. In the latter case, `recv_bytes()` waits for that false length and the parent stops forwarding worker logs.

## Affected versions

Reproduced with:

- Dramatiq 1.17.1
- Dramatiq 2.2.0 (current latest release)

## Reproduction

Use five concurrent writers sharing one `StreamablePipe`, each writing 128 KiB JSON records. With 500 records per writer, the unmodified 2.2.0 implementation consistently reports frame corruption, for example:

```text
frame corruption after 0 of 2500 frames: UnicodeDecodeError
```

Serializing the same writes with a process-local lock delivers all frames:

```text
received all 2500 frames
```

The relevant current implementation is:

```python
def write(self, s):
self.pipe.send_bytes(s.encode(self.encoding, errors="replace"))
```

## Proposed fix

Add a process-local lock to `StreamablePipe` and hold it around the complete `send_bytes()` call. The lock must be recreated on unpickle so spawned worker processes remain supported.

```python
with self._write_lock:
self.pipe.send_bytes(s.encode(self.encoding, errors="replace"))
```

A regression test can use two threads and a fake `send_bytes()` implementation that detects overlapping calls, plus a pickle test for `StreamablePipe`.

I have a local patch and focused regression test available if a PR would be useful.

Look at the StreamablePipe class in the codebase, likely in a file related to logging or multiprocessing. Understand how write() currently calls send_bytes. The fix involves adding a process-local lock around the send_bytes call and ensuring it works after unpickling. Write a regression test using threads to simulate concurrent writes and verify no corruption occurs.

Scritto dal modello di indicizzazione a partire dal testo della issue.