[Rust] Fix Arrow failed-enqueue finalization

[nikolaobradovic-db]

What changes are proposed in this pull request?

• Roll back Arrow logical offsets and record ranges when enqueueing fails with recovery disabled.
• Claim terminal admission before releasing the ingest lock, preventing concurrent close from masking the request-stream error.
• Await shared terminal finalization instead of racing a fixed local timeout.
• Serialize flush target snapshots with enqueue rollback so withdrawn offsets are never awaited.
• Add coverage for first-batch failure, acknowledged-prefix rollback, delayed finalization, and concurrent close.

How is this tested?

• New concurrency regression test verified red without the admission claim and green with it.

[teodordelibasic-db]

Should we bound the wait with flush_timeout_ms?

return match timeout(
    Duration::from_millis(self.options.flush_timeout_ms),
    self.wait_for_terminal_outcome(),
)
.await
{
    Err(_) => Err(ZerobusError::StreamClosedError(
        tonic::Status::deadline_exceeded("Failed to send batch"),
    )),
    Ok(Err(error)) => Err(error),
    Ok(Ok(())) => Err(Self::terminal_error_or(&self.server_error_rx, || {
        "Failed to send batch".to_string()
    })),
};

wait_for_offset_internal is bounded by flush_timeout_ms, but this path loops until CloseState::Finalized. The empty-target flush() branch and the Open plus admission_closed arm of close_internal have the same unbounded wait. If the supervisor never reaches finish(), ingest_batch hangs a caller ingest loop.

The rustdoc says this call returns the request-stream error, but that is only true on Err. The Ok(()) arm synthesizes Status::internal("Failed to send batch"), while the supervisor's send failure is Status::unavailable with Flight request stream closed while sending. That Ok arm looks unreachable after the admission_closed claim, but if we keep it, could we still surface whatever is on server_error_rx instead of a generic internal?