[Rust][Arrow] Move explicit close into supervisor

[teodordelibasic-db]

What changes are proposed in this pull request?

Arrow explicit close previously depended on caller-side flush completion before supervisor teardown could begin. During recovery, that allowed backoff, reconnect setup, ready-signal waiting, credential invalidation, or replay to continue until a separate timeout resolved the phase.

This PR moves explicit-close coordination and finalization into the Arrow stream supervisor. The close target and original deadline are published together while holding the ingestion mutex, and the supervisor owns the single teardown.

• Interrupt recovery backoff, reconnect setup, ready-signal waiting, and partial replay when close is requested. Explicit close does not start or continue recovery.
• Keep active acknowledgment processing alive through the close flush, then half-close the request and use a fresh bounded drain window even if the flush deadline has expired. A concrete peer error found during that drain takes precedence over a synthetic flush timeout.
• Skip the remaining flush wait after the active transport has already failed, while still half-closing and draining the response for a concrete peer error.
• Linearize close publication and terminal outcome selection under the ingestion mutex so an acknowledged close target wins over a peer error observed in the same response cycle.
• Preserve reconnect timeout and recovery-trigger error precedence while allowing authentication rejection to remain the concrete terminal outcome.
• Run reconnect credential invalidation in an independently owned, bounded task, including when close was already published before the authentication rejection became visible.
• Schedule bounded terminal credential invalidation before finalization and detach it from the supervisor lifecycle so close or stream drop cannot cancel it.
• Publish a replacement sender only after reconnect replay bookkeeping is ready, so cancellation cannot expose a sender whose transport has already been dropped.
• Preserve a peer error already received by a server-rotation drain when close begins concurrently.
• Keep close inside an in-progress server-rotation ACK wait, transfer ownership of the published close request into that rotation's drain, and preserve its submitted snapshot and deadlines. Latch a close deadline throughout the ACK wait and drain, including when the biased response path wins a deadline tie or ACK application yields, so a late ACK cannot retroactively turn the timed-out close into success.
• When server rotation begins during the explicit-close flush wait, let that rotation own the single half-close and response drain, then finalize directly from its outcome.
• Retain only the exact unacknowledged batch suffixes when close interrupts partial replay.
• Keep repeated close calls idempotent with the original deadline and stored outcome. Polling close() publishes an irreversible close request; if that future is cancelled, ingestion remains closed and callers must invoke close() again to await completion.
• Finalize locally with a terminal error if the supervisor exits abnormally after close is requested, so close() cannot hang and unacknowledged batches remain retrievable.
• Store the selected outcome when close enters finalization so abnormal-exit fallback returns that exact result rather than recomputing it from the wall clock, ACK watermark, or an older server error. Keep the published server error synchronized with that selected outcome.
• Serialize finalization with ingestion so accepted batches are included in the final failed-batch snapshot.

This is an internal Rust SDK lifecycle change. It does not change public APIs, FFI signatures, ABI, or semver compatibility.

Remaining intentional limitations:

• Cleanup of a reconnect transport that does not complete setup or replay remains best-effort.
• Concurrent ingest and close across an FFI boundary remains unsupported.
• When a prior transport's durable outcome is unknowable, replay on a replacement stream retains unavoidable cross-stream ambiguity even though the client preserves the exact locally unacknowledged suffix.

Fixes #657.

How is this tested?

• test_close_publication_preserves_target_ack_precedence: publishes close while terminal finalization is parked after a target ACK, aborts the supervisor after outcome selection, and verifies that fallback preserves the acknowledged-target result over the following peer error.
• test_close_does_not_cancel_received_ack_application: verifies close lets a consumed ACK finish applying before selecting the terminal outcome.
• test_close_finalizes_after_supervisor_abort: verifies abnormal supervisor exit cannot leave close hanging and preserves the unacknowledged batch for retrieval.
• test_supervisor_abort_preserves_post_deadline_timeout: applies a late ACK during post-deadline drain, aborts after the timeout outcome is selected, and verifies fallback preserves that timeout while retaining the ACK for pending-batch accounting.
• test_close_skips_flush_wait_after_request_send_failure: verifies an already-failed active transport bypasses the remaining flush wait and that a retryable post-EOF drain error does not mask the original request-send failure.
• test_expired_close_timeout_drains_post_eof_error: verifies an expired flush deadline still receives a bounded post-EOF drain and preserves a concrete peer rejection over the flush timeout.
• test_expired_close_timeout_applies_post_eof_ack_without_succeeding: verifies a post-EOF ACK still clears acknowledged pending work after the flush deadline without retroactively turning the timed-out close into success.
• test_close_during_rotation_drain_preserves_peer_error: verifies close does not lose a permanent peer status already received by an in-progress rotation drain.
• close_during_rotation_ack_wait_preserves_rotation_state: verifies target-already-acknowledged, deadline-expired, and target-acknowledged-after-publication closes remain in the active rotation state, including a deterministic tie where the ACK and close deadline are both ready.
• deadline_driven_close_drain_reports_flush_timeout: verifies a deadline-driven rotation drain retains the close flush-timeout classification.
• test_rotation_close_timeout_survives_late_drain_ack: uses a positive close timeout and an absolute post-publication deadline while parking target-ACK application during rotation drain, then verifies close reports the flush timeout while retaining the ACK for pending-batch accounting.
• test_close_flush_overlapping_rotation_drains_once: publishes close before a delayed rotation signal, then verifies the nested flush path completes from that rotation's post-EOF ACK without panicking, reconnecting, or draining the transport twice.
• test_close_after_reconnect_timeout_preserves_timeout_error: publishes close immediately after reconnect timeout wins and verifies the timeout remains the terminal outcome.
• test_published_close_preserves_recovery_trigger_over_retryable_reconnect_failure: verifies a teardown-induced retryable reconnect failure does not mask the error that started recovery.
• test_published_close_still_invalidates_rejected_reconnect_credentials: verifies an authentication rejection invalidates cached credentials even when close was already published.
• test_close_preserves_reconnect_auth_rejection_while_invalidation_completes: verifies close returns the authentication rejection promptly while the independently owned invalidation finishes afterward.
• test_close_does_not_cancel_terminal_auth_invalidation: races close with an active authentication rejection and verifies the single invalidation survives both finalization and stream drop.
• cancelled_reconnect_commit_does_not_publish_sender: verifies cancellation while reconnect commit is waiting cannot publish a replacement sender or unpause ingestion.
• test_close_interrupts_recovery_backoff: verifies close interrupts a parked recovery backoff and preserves the triggering error.
• test_close_interrupts_reconnect_setup_and_ready_wait: verifies close interrupts both a blocked reconnect handshake and ready-signal wait.
• test_close_during_reconnect_rebuild_and_partial_replay_retains_exact_unacked_suffixes: verifies close retains the exact durable-watermark suffix both before range rebasing and during partial replay.
• test_cancelled_close_rejects_ingest_and_resumes_teardown: verifies cancelling the first close future leaves the request active and a repeated call resumes with the original deadline and result.
• test_empty_stream_close_preserves_terminal_error: verifies closing an empty stream preserves an existing terminal error.
• test_idempotent_close: verifies a clean empty stream with a zero close deadline closes successfully and repeated close returns the stored outcome.
• test_unrepresentable_ack_and_flush_timeouts_are_rejected: verifies stream creation rejects ACK and flush timeouts that cannot form platform deadlines.
• test_unrepresentable_runtime_recovery_deadline_is_rejected: verifies recovery reports an invalid timeout configuration without masking the error that triggered recovery during close.
• pause_and_detach_waits_for_in_flight_ingest: verifies sender detachment waits for an in-flight ingestion critical section.
• finalize_closed_waits_for_in_flight_ingest: verifies finalization includes a batch accepted by in-flight ingestion.