receiver: wait for batched FDs instead of returning MismatchedCount

[giuseppe]

When the sender batches file descriptors across multiple sendmsg() calls, the receiver may parse the complete JSON message before all FDs have arrived. Previously this returned a MismatchedCount error immediately.

Buffer the parsed message in a pending_message field and let the receive() loop continue calling read_more_data() until enough FDs have been collected.

Assisted-by: Claude Opus 4.6 noreply@anthropic.com

@cgwalters does this make any sense? The issue was found by Claude and it is required to use the crate in composefs/composefs-rs#228

[gemini-code-assist]

Code Review

This pull request enhances the UnixSocketTransport to handle batched file descriptors by introducing a pending_message buffer in the Receiver. This prevents MismatchedCount errors when JSON messages are parsed before all associated FDs have arrived. The review feedback suggests optimizing the retrieval of pending messages using Option::take to avoid expensive clones and recommends refactoring duplicated message construction logic into a private helper method.

[gemini-code-assist]

To improve efficiency, you can avoid cloning the serde_json::Value here. By using self.pending_message.take(), you can move the value out of the Option. If you find that there aren't enough file descriptors yet, you can move it back. This avoids a potentially expensive allocation and deep copy of the JSON value.

        if let Some((value, fd_count)) = self.pending_message.take() {
            if self.fd_queue.len() >= fd_count {
                let fds: Vec<OwnedFd> = (0..fd_count)
                    .map(|_| self.fd_queue.pop_front().unwrap())
                    .collect();

                let message = JsonRpcMessage::from_json_value(value)?;
                return Ok(Some(MessageWithFds::new(message, fds)));
            }
            // Not enough FDs, put it back and wait for more.
            self.pending_message = Some((value, fd_count));
            return Ok(None);
        }

[gemini-code-assist]

This block of code for creating a MessageWithFds from a value and fd_count is very similar to the code on lines 358-363. To improve maintainability and follow the DRY (Don't Repeat Yourself) principle, consider extracting this logic into a private helper method within impl Receiver.

[cgwalters]

OK yes sorry, this was buggy...I did a PR for the spec here #10

[cgwalters]

Assisted-by: OpenCode (Claude Opus 4)

AI-generated review. Comments prefixed with AI: are unedited.

[cgwalters]

AI: Important: When the connection closes while pending_message is Some, read_more_data() returns Error::ConnectionClosed and the pending message is silently abandoned. The caller gets ConnectionClosed instead of MismatchedCount, losing diagnostic info about expected vs. found FD counts.

Worse, receive_opt() converts ConnectionClosed into Ok(None), making a protocol error look like a clean shutdown.

Suggest intercepting this in receive():

pub async fn receive(&mut self) -> Result<MessageWithFds> {
    loop {
        if let Some(message) = self.try_parse_message()? {
            return Ok(message);
        }
        match self.read_more_data().await {
            Err(Error::ConnectionClosed) if self.pending_message.is_some() => {
                let (_, fd_count) = self.pending_message.take().unwrap();
                return Err(Error::MismatchedCount {
                    expected: fd_count,
                    found: self.fd_queue.len(),
                });
            }
            Err(e) => return Err(e),
            Ok(()) => {}
        }
    }
}

This preserves the "wait for batched FDs" behavior while restoring the diagnostic when the connection actually dies. It also means receive_opt() correctly propagates MismatchedCount as an error rather than swallowing it as Ok(None).

The existing tests test_mismatched_fd_count_error and test_fds_field_mismatch_too_few_fds should also be tightened to assert on the specific error variant.

[cgwalters]

AI: (low) Worth noting that while a message is pending, all subsequent message parsing is blocked — even messages needing 0 FDs. This is correct for maintaining FIFO ordering on a Unix socket, but a brief comment documenting this serialization property would help future readers.

[cgwalters]

AI: (low) Per REVIEW.md: "Avoid the if let Ok(v) = ... { } pattern by default." If accept() fails, the test body is silently skipped and the test passes vacuously. Consider .unwrap() instead. (Pre-existing pattern in other tests, but worth not propagating further.)

[cgwalters]

I think this would be more elegant with fd_queue.drain()

[cgwalters]

i think with Rust 2024 edition we could fold this conditional into the above and avoid a take -> put cycle

[cgwalters]

Needs DCO

[cgwalters]

I think we can also avoid an is_some() + unwrap() here by adding to the match above

[cgwalters]

Minor nit i'd downgrade this to trace