fix(core): retain buffered data in WriteGenerator on write failure

[TennyZhuang]

Which issue does this PR close?

N/A - discovered via deterministic testing of error-handling paths.

Rationale

WriteGenerator's write() and close() methods called buffer.take() before attempting to write to the underlying writer. If the write failed, the buffered data was permanently lost, making retry impossible without data loss.

This violated the invariant maintained by other writer implementations (AppendWriter, BlockWriter, MultipartWriter, PositionWriter), which only clear state after a successful write.

Changes

• Inexact mode write path: Clone the taken buffer before writing; restore it on failure.
• Exact mode write path: Same clone-and-restore pattern.
• close() method: Same clone-and-restore pattern in the flush loop.
• Added 3 deterministic tests: FailOnceMockWriter that fails the first write call, then succeeds — verifying no data loss in inexact mode, exact mode, and close() paths.

Are there any user-facing changes?

Users who rely on retry (e.g., via RetryLayer) will no longer silently lose data when the underlying storage temporarily fails during a write flush.

This PR was generated with the assistance of AI (Claude).

[dentiny]

May I ask some dumb questions?

• It seems to me that the "issue" reproduces if users call write again after a failed write. In production, why would users do that?
• In terms of retry layer (which calls write repeatedly after failure), it seems to sit below the write generator. I'm curious if we could provide a reproducible example to show data loss with retry layer used together?

[TennyZhuang]

@dentiny Good question. A few clarifications:

1. RetryWrapper has a limited retry budget. While RetryWrapper does clone the buffer on each internal retry attempt (r.write(bs.clone()).await), once all retries are exhausted, the error propagates back to WriteGenerator. The buffer passed to RetryWrapper is consumed (moved), so WriteGenerator's internal buffer (taken via buffer.take()) is gone.

2. Application-level retry is common. Even with RetryLayer configured, users commonly add their own retry logic at the application level. For example:

   for _ in 0..max_retries {
       match writer.write(data).await {
           Ok(()) => break,
           Err(e) if e.is_temporary() => continue,
           Err(e) => return Err(e),
       }
   }

   In this pattern, if write(data) fails during a flush (because RetryWrapper exhausted its retries), the user retries with the same data. But the previously buffered data from an earlier write() call is no longer in the argument — it was already consumed from WriteGenerator's buffer.

3. Real-world evidence — RisingWave's foyer project (mentioned in the PR thread, foyer: error handling improvement risingwavelabs/risingwave#24673) hit this exact issue. Batch processing systems frequently retry failed write operations at the task level.

I've added a test test_inexact_write_failure_with_retry_style_retry on the branch that demonstrates this scenario step by step.