Fix a potential bug in DropCloserReadHalf::take()

[steedmicro]

Description

I fixed a potential bug in DropCloserReadHalf::take() function.
I mainly applied @allada's recommendations.
Though, I kept the previous second_chunk logic because we need to check second_chunk is empty or not in order to return first_chunk without copying stuff.
Detailed context can be found in the issue below.

Fixes #578

Type of change

• Bug fix (non-breaking change which fixes an issue)

Checklist

• Updated documentation if needed
• Tests added/amended
• bazel test //... passes locally
• PR is contained in a single commit, using git amend see some docs

---

This change is 

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

Name	Status	Preview	Comments	Updated (UTC)
nativelink-docs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Jan 12, 2024 1:07pm

[allada]

Reviewable status: 0 of 1 LGTMs obtained

---

nativelink-util/src/buf_channel.rs line 321 at r1 (raw file):

            // 3a. If our `first_chunk` is EOF, we are done.
            // 3b. If our `first_chunk` is exactly `size` we can return our current state.
            if first_chunk.is_empty() || first_chunk.len() >= size {

I don't think this part is correct.

We need to receive the EOF, and if the first chunk returns exactly the number of bytes we won't receive our EOF.

If first_chunk.is_empty() is true this is fine.
If self.partial.is_some() is true, it is ok to go down this path but we should assert first_chunk.len() == size.

Id rewrite this as:

// 3a. If our `first_chunk` is EOF, we are done.
if first_chunk.is_empty() {
  return Ok(first_chunk);
}

// 3b. If our first_chunk has data and it our self.partial was filled it means our stream has more data.
if self.partial.is_some() {
  assert!(first_chunk.len() == size, "Length should be exactly size here");
  return Ok(first_chunk);
}

---

nativelink-util/src/buf_channel.rs line 364 at r1 (raw file):

            output.put(chunk);

            if output.len() == size {

We can only do this if output.len() == size && self.partial.is_some()

---

nativelink-util/src/buf_channel.rs line 367 at r1 (raw file):

                return Ok(output.freeze());
            }
            assert!(output.len() < size, "Length should never be larger than size in take()");

nit: can needs to be <=

---

nativelink-util/src/buf_channel.rs line 369 at r1 (raw file):

            assert!(output.len() < size, "Length should never be larger than size in take()");
            assert!(
                self.partial.is_none(),

nit: We can remove this, it's checked at the beginning of the loop.