diff --git a/fluss-rust/bindings/cpp/BUILD.bazel b/fluss-rust/bindings/cpp/BUILD.bazel index c751266fe63..4f01fc6781d 100644 --- a/fluss-rust/bindings/cpp/BUILD.bazel +++ b/fluss-rust/bindings/cpp/BUILD.bazel @@ -55,6 +55,9 @@ genrule( name = "cargo_build_debug", srcs = glob([ "src/**/*.rs", + "src/**/*.hpp", + "include/**/*.hpp", + "build.rs", "Cargo.toml", ]), outs = [ @@ -121,6 +124,9 @@ genrule( name = "cargo_build_release", srcs = glob([ "src/**/*.rs", + "src/**/*.hpp", + "include/**/*.hpp", + "build.rs", "Cargo.toml", ]), outs = [ @@ -274,6 +280,7 @@ cc_library( textual_hdrs = [ "src/ffi_converter.hpp", "src/type_lowering.hpp", + "src/write_callback.hpp", ":rust_bridge_h_unified", ":lib_rs_h_unified", ":cxx_h_unified", diff --git a/fluss-rust/bindings/cpp/README.md b/fluss-rust/bindings/cpp/README.md index 35c2ecd2771..00dc11a1704 100644 --- a/fluss-rust/bindings/cpp/README.md +++ b/fluss-rust/bindings/cpp/README.md @@ -65,7 +65,7 @@ not apply to `CreateBucketBatchScanner()`. ## Examples and Documentation -- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes, continuous scans, +- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes with Wait and bounded callbacks, continuous scans, bounded Arrow record-batch scans, projections, and offset queries. - [examples/admin_example.cpp](examples/admin_example.cpp) demonstrates database, table, partition, and cluster administration. @@ -76,6 +76,40 @@ not apply to `CreateBucketBatchScanner()`. [C++ API reference](../../website/docs/user-guide/cpp/api-reference.md) and [log-table examples](../../website/docs/user-guide/cpp/example/log-tables.md). +The SDK executes `WriteCallback` (`void(const WriteCompletion&)`) on one shared +worker, serially in dispatch order and off the I/O threads. `WriteCompletion.result` +is the write outcome; copy it before passing it to another worker. +`CreateWriter(writer)` uses the default `WriteCallbackOptions`; the overload +`CreateWriter(writer, options)` accepts a positive `max_pending_operations` limit +(default 262144) per writer. This operation-count budget is independent of the +Connection's byte-counted write buffer. Slow callbacks can fill it even when the +write buffer has room. The Rust write-buffer permit is released when the batch +completes, before the user callback returns, but the callback object and its +captures remain retained until callback completion. Once the per-writer callback +limit is full, callback-based submissions wait or fail according to +`client.writer.buffer.wait-timeout`; its default is unbounded. + +Callback-capacity and buffer waits share `client.writer.buffer.wait-timeout`. +Zero makes those waits fail fast; this is not a deadline for the entire API call, +ACKs, retries, or callback execution. See the +[buffer sizing guidance](../../website/docs/user-guide/cpp/api-reference.md#sizing-callback-capacity-and-write-buffers) +for independent capacity and byte budgets. Callback worker initialization failure +rejects the submission before any data is accepted; there is no parallel fallback. + +A failed callback does not prove that the record was not written. Application +resubmission can duplicate it, even with SDK idempotence enabled. The example only +counts and logs outcomes; it does not implement durable recovery. Keep callbacks +short, protect shared state, and handle retries outside the callback with an +application recovery policy. + +After submissions stop, `Flush()` first flushes writes and, on success, blocks until +pending callbacks finish, acting as a barrier. A callback that never returns hangs it. +Calling it from a write callback is rejected before flushing. If a write flush +returns an error, keep callback state alive; if +it succeeds, still check individual write results. +See the [callback guarantees and recovery guidance](../../website/docs/user-guide/cpp/api-reference.md#write-guarantees-and-recovery) +for result semantics, callback implementation, and shutdown requirements. + For a bounded log scan, pass the per-bucket offset ranges directly to `TableScan`. The returned reader yields one Arrow batch at a time until every `[starting_offset, stopping_offset)` range is complete: diff --git a/fluss-rust/bindings/cpp/build.rs b/fluss-rust/bindings/cpp/build.rs index ec75e24aebd..7a633c606b2 100644 --- a/fluss-rust/bindings/cpp/build.rs +++ b/fluss-rust/bindings/cpp/build.rs @@ -17,8 +17,12 @@ fn main() { cxx_build::bridge("src/lib.rs") + .include("include") + .include("src") .std("c++17") .compile("fluss-cpp-bridge"); println!("cargo:rerun-if-changed=src/lib.rs"); + println!("cargo:rerun-if-changed=src/write_callback.hpp"); + println!("cargo:rerun-if-changed=include/fluss.hpp"); } diff --git a/fluss-rust/bindings/cpp/examples/example.cpp b/fluss-rust/bindings/cpp/examples/example.cpp index 14a8380f2a0..08e77a64a67 100644 --- a/fluss-rust/bindings/cpp/examples/example.cpp +++ b/fluss-rust/bindings/cpp/examples/example.cpp @@ -20,9 +20,13 @@ #include #include +#include #include #include +#include +#include #include +#include #include #include "fluss.hpp" @@ -39,6 +43,10 @@ int main() { // 1) Connect fluss::Configuration config; config.bootstrap_servers = "127.0.0.1:9123"; + // Callback capacity and buffer waits share this budget. The default UINT64_MAX + // waits indefinitely; a finite value allows handling overload instead. Zero fails + // fast when either resource is unavailable. This is not a whole-call deadline. + config.writer_buffer_wait_timeout_ms = 30000; fluss::Connection conn; check("create", fluss::Connection::Create(config, conn)); @@ -85,7 +93,11 @@ int main() { // 5) Write rows with scalar and temporal values fluss::AppendWriter writer; - check("new_append_writer", table.NewAppend().CreateWriter(writer)); + fluss::WriteCallbackOptions callback_options; + // Per writer, independent of config.writer_buffer_memory_size (per Connection). + // CreateWriter(writer) without options uses the default 262144-operation limit. + callback_options.max_pending_operations = 4096; + check("new_append_writer", table.NewAppend().CreateWriter(writer, callback_options)); struct RowData { int id; @@ -143,6 +155,75 @@ int main() { std::cout << "Row acknowledged by server" << std::endl; } + // Callback acknowledgment + { + // The SDK runs callbacks; no application waiting thread is required. + // Callbacks run on one shared worker, so keep them short and + // non-blocking: do not Flush/Wait or retry synchronously inside a callback. + // This example counts outcomes only. It does not implement durable recovery. + struct CallbackState { + std::atomic succeeded{0}; + std::atomic failed{0}; + std::mutex mutex; + int32_t first_failed_id{0}; + fluss::Result first_failure; + }; + // Shared ownership also keeps state alive if submission throws or flushing fails. + auto state = std::make_shared(); + bool submission_failed = false; + for (const auto& r : rows) { + const int32_t id = 1000 + r.id; + fluss::GenericRow row; + row.SetInt32(0, id); + row.SetString(1, r.name); + row.SetFloat32(2, r.score); + row.SetInt32(3, r.age); + row.SetDate(4, r.date); + row.SetTime(5, r.time); + row.SetTimestampNtz(6, r.ts_ntz); + row.SetTimestampLtz(7, r.ts_ltz); + auto submitted = writer.Append( + row, [id, state](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + if (result.Ok()) { + ++state->succeeded; + } else { + // Copy the invocation-scoped result. Do not log, perform I/O, + // or retry here: one slow callback delays every writer. + if (state->failed.fetch_add(1) == 0) { + std::lock_guard lock(state->mutex); + state->first_failed_id = id; + state->first_failure = result; + } + } + }); + if (!submitted.Ok()) { + // No callback will run for this submission; handle this path too. + submission_failed = true; + std::cerr << "Submission failed for id=" << id << ": " << submitted.error_message + << '\n'; + break; + } + } + // Stop submissions, then drain accepted callbacks. Flush success alone does + // not mean every write succeeded. Writer/Connection destruction is not a drain. + check("flush", writer.Flush()); + std::cout << "Callback writes: succeeded=" << state->succeeded.load() + << " failed=" << state->failed.load() << '\n'; + if (state->failed.load() != 0) { + std::lock_guard lock(state->mutex); + std::cerr << "First failed id=" << state->first_failed_id + << ": " << state->first_failure.error_message << '\n'; + } + // A failed write may have reached the server. Recover outside the callback + // using retained input or a replayable source and application-level deduplication. + // Advance a source position only after Flush and all relevant writes succeed. + // This example reports failure and exits; it does not implement durable recovery. + if (submission_failed || state->failed.load() != 0) { + return 1; + } + } + // Append a row with all fields null (matches Rust log_table.rs all_supported_datatypes) { fluss::GenericRow row; diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp index d9799e0912a..fff1b850773 100644 --- a/fluss-rust/bindings/cpp/include/fluss.hpp +++ b/fluss-rust/bindings/cpp/include/fluss.hpp @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -47,6 +48,8 @@ struct Admin; struct Table; struct AppendWriter; struct WriteResult; +class WriteCallback; +class WriteCallbackCapacity; struct LogScanner; struct RecordBatchLogReader; struct BatchScanner; @@ -527,10 +530,62 @@ struct Result { bool Ok() const { return error_code == 0; } - /// Returns true if retrying the request may succeed. Client-side errors always return false. + /// Returns true if retrying the request may succeed. Does not guarantee that a failed + /// write had no effect or that application resubmission is duplicate-safe. + /// Client-side errors always return false. bool IsRetriable() const { return ErrorCode::IsRetriable(error_code); } }; +/// Write-specific completion metadata. The reference passed to a callback is valid +/// only for that invocation; copy the result when retaining it for later work. +struct WriteCompletion { + Result result; +}; + +/// Per-writer admission control for callback operations, independent of buffer bytes. +struct WriteCallbackOptions { + // Includes accepted writes awaiting completion and callbacks queued or executing. + // Must be greater than zero. This is not a byte limit on callback captures. + size_t max_pending_operations{262144}; +}; + +/// Receives the final outcome of an accepted write. Function pointers and lambdas +/// are supported. An empty callback is rejected before submitting the write. +/// +/// During normal operation, the SDK owns the callback until completion and invokes +/// it exactly once on a shared SDK worker; no caller polling or waiting thread +/// is needed. Process exit or a crash can prevent delivery. Callbacks never run +/// inline in the submitting call, but may start before the call returns. +/// Keep callbacks short and synchronize access to shared state, including writers. +/// Callback overloads do not make writers safe for concurrent access. Captured +/// references must remain valid until the callback finishes; capturing shared +/// ownership is recommended. Keep the connection alive until completion. +/// +/// Success follows the configured acknowledgment policy. Errors are reported after +/// internal retry handling, but do not guarantee that no data was written. +/// Application resubmission is a new operation and can produce duplicates even +/// with SDK idempotence enabled. Retain input identifiers and recovery state as +/// needed; one callback invocation is not an exactly-once delivery guarantee. +/// +/// Callbacks run serially in dispatch order on a single shared worker. There is +/// no cross-bucket submission-order guarantee; late registrations are queued +/// when registered. Do not wait for another callback from a callback: it stalls that +/// worker. Synchronous SDK calls require exclusive writer access. Callback +/// submissions to a full writer fail immediately when called from a callback, +/// instead of blocking the worker. Each writer bounds its outstanding callback +/// operations using WriteCallbackOptions, independently of the write buffer size. +/// Hand off retries or expensive work without blocking; bound application queues +/// and handle overflow without silently discarding failed operations. +/// +/// Exceptions thrown by callbacks are caught and reported to stderr; they do not +/// change the write outcome or retry the callback. Stop submissions before Flush(). +/// After a successful Rust write flush, Flush() blocks until pending callbacks finish, +/// acting as a barrier, so a callback that never returns hangs it. On error, referenced +/// state may still be in use. +/// Flush() called inside any write callback returns a client error without flushing. +/// Flush() does not wait for work handed to application workers or retry queues. +using WriteCallback = std::function; + struct TablePath { std::string database_name; std::string table_name; @@ -1553,9 +1608,12 @@ struct Configuration { bool writer_enable_idempotence{true}; // Maximum number of in-flight requests per bucket for idempotent writes size_t writer_max_inflight_requests_per_bucket{5}; - // Total memory available for buffering write batches (default 64MB) + // Shared write-batch memory budget per Connection, across its tables and writers + // (default 64 MiB). Not a process RSS limit or a callback-capture memory budget. size_t writer_buffer_memory_size{64 * 1024 * 1024}; - // Maximum time in milliseconds to block waiting for buffer memory + // Shared wait budget in milliseconds for buffer memory and callback capacity. + // Does not bound data conversion, scheduling, ACKs, or callback execution. + // UINT64_MAX waits indefinitely; zero fails fast when capacity or memory is unavailable. uint64_t writer_buffer_wait_timeout_ms{std::numeric_limits::max()}; // Maximum KV backpressure throttle in milliseconds uint64_t writer_kv_backpressure_max_throttle_ms{3000}; @@ -1740,6 +1798,8 @@ class TableAppend { TableAppend& operator=(TableAppend&&) noexcept = default; Result CreateWriter(AppendWriter& out); + /// Create a writer with an independent, positive callback operation limit. + Result CreateWriter(AppendWriter& out, const WriteCallbackOptions& options); private: friend class Table; @@ -1759,6 +1819,8 @@ class TableUpsert { TableUpsert& PartialUpdateByName(std::vector column_names); Result CreateWriter(UpsertWriter& out); + /// Create a writer sharing one callback operation limit across upserts and deletes. + Result CreateWriter(UpsertWriter& out, const WriteCallbackOptions& options); private: friend class Table; @@ -1877,6 +1939,7 @@ class WriteResult { friend class UpsertWriter; WriteResult(ffi::WriteResult* inner) noexcept; + Result Notify(std::unique_ptr callback); void Destroy() noexcept; ffi::WriteResult* inner_{nullptr}; }; @@ -1895,17 +1958,37 @@ class AppendWriter { Result Append(const GenericRow& row); Result Append(const GenericRow& row, WriteResult& out); + /// Submit a row and notify callback of its final outcome without waiting for + /// acknowledgment. Callback capacity and buffer waits share the budget from + /// client.writer.buffer.wait-timeout. Ok means the write was accepted and the + /// callback fires exactly once during normal operation; an error means + /// submission failed and no callback runs. A zero timeout makes admission + /// fail fast when callback capacity or buffer memory is unavailable. + Result Append(const GenericRow& row, WriteCallback callback); Result AppendArrowBatch(const std::shared_ptr& batch); Result AppendArrowBatch(const std::shared_ptr& batch, WriteResult& out); + /// Like the callback Append overload, but notifies once for the entire batch. + Result AppendArrowBatch(const std::shared_ptr& batch, + WriteCallback callback); Result Flush(); private: friend class Table; friend class TableAppend; - AppendWriter(ffi::AppendWriter* writer) noexcept; + AppendWriter(ffi::AppendWriter* writer, + std::shared_ptr callback_capacity) noexcept; + + // Submit through FFI bounding the buffer-backpressure wait by submit_budget_ms + // (Kafka max.block.ms style): negative uses the writer's configured buffer wait + // timeout, >= 0 caps the wait at that many ms (0 = fail fast when the buffer is + // full). Only the callback path passes a budget; the public overloads pass -1. + Result AppendWithBudget(const GenericRow& row, WriteResult& out, int64_t submit_budget_ms); + Result AppendArrowBatchWithBudget(const std::shared_ptr& batch, + WriteResult& out, int64_t submit_budget_ms); void Destroy() noexcept; ffi::AppendWriter* writer_{nullptr}; + std::shared_ptr callback_capacity_; }; class UpsertWriter { @@ -1922,16 +2005,33 @@ class UpsertWriter { Result Upsert(const GenericRow& row); Result Upsert(const GenericRow& row, WriteResult& out); + /// Submit an upsert and notify callback of its final outcome without waiting + /// for acknowledgment. Callback capacity and buffer waits share the budget from + /// client.writer.buffer.wait-timeout. Ok means the write was accepted and the + /// callback fires exactly once during normal operation; an error means + /// submission failed and no callback runs. A zero timeout makes admission + /// fail fast when callback capacity or buffer memory is unavailable. + Result Upsert(const GenericRow& row, WriteCallback callback); Result Delete(const GenericRow& row); Result Delete(const GenericRow& row, WriteResult& out); + /// Like the callback Upsert overload, but deletes a row by primary key. + Result Delete(const GenericRow& row, WriteCallback callback); Result Flush(); private: friend class Table; friend class TableUpsert; - UpsertWriter(ffi::UpsertWriter* writer) noexcept; + UpsertWriter(ffi::UpsertWriter* writer, + std::shared_ptr callback_capacity) noexcept; + + // See AppendWriter::AppendWithBudget for submit_budget_ms semantics. Only the + // callback path passes a budget; the public overloads pass -1. + Result UpsertWithBudget(const GenericRow& row, WriteResult& out, int64_t submit_budget_ms); + Result DeleteWithBudget(const GenericRow& row, WriteResult& out, int64_t submit_budget_ms); + void Destroy() noexcept; ffi::UpsertWriter* writer_{nullptr}; + std::shared_ptr callback_capacity_; }; class Lookuper { diff --git a/fluss-rust/bindings/cpp/src/lib.rs b/fluss-rust/bindings/cpp/src/lib.rs index b9ca1f93b95..04d6945cfc1 100644 --- a/fluss-rust/bindings/cpp/src/lib.rs +++ b/fluss-rust/bindings/cpp/src/lib.rs @@ -16,6 +16,8 @@ // under the License. mod types; +mod write_callback; +use write_callback::ensure_callback_executor; use std::collections::HashMap; use std::str::FromStr; @@ -42,6 +44,15 @@ static RUNTIME: LazyLock = LazyLock::new(|| { #[cxx::bridge(namespace = "fluss::ffi")] mod ffi { + unsafe extern "C++" { + include!("write_callback.hpp"); + + type WriteCallback; + + #[cxx_name = "Complete"] + fn complete(self: Pin<&mut WriteCallback>, error_code: i32, error_message: &str); + } + struct HashMapValue { key: String, value: String, @@ -521,6 +532,8 @@ mod ffi { unsafe fn get_arrow_schema(self: &Table, out_ptr: usize) -> FfiResult; fn get_table_path(self: &Table) -> FfiTablePath; fn has_primary_key(self: &Table) -> bool; + fn writer_buffer_wait_timeout_ms(self: &Table) -> u64; + fn ensure_callback_executor() -> FfiResult; fn create_upsert_writer(self: &Table, column_indices: Vec) -> FfiPtrResult; fn new_lookuper(self: &Table) -> FfiPtrResult; fn new_prefix_lookuper(self: &Table, lookup_column_names: Vec) -> FfiPtrResult; @@ -657,24 +670,33 @@ mod ffi { // AppendWriter unsafe fn delete_append_writer(writer: *mut AppendWriter); - fn append(self: &mut AppendWriter, row: &GenericRowInner) -> FfiPtrResult; + // budget_ms bounds the buffer-memory wait: negative uses the writer's + // configured buffer wait timeout, >= 0 caps the wait at that many ms + // (0 = non-blocking), letting callers keep a submit within a fixed budget. + fn append(self: &mut AppendWriter, row: &GenericRowInner, budget_ms: i64) -> FfiPtrResult; // Partition (if partitioned) comes from the first row, so all rows must // share one partition; rows are distributed across buckets by key. fn append_arrow_batch( self: &mut AppendWriter, array_ptr: usize, schema_ptr: usize, + budget_ms: i64, ) -> FfiPtrResult; fn flush(self: &mut AppendWriter) -> FfiResult; // WriteResult unsafe fn delete_write_result(wr: *mut WriteResult); fn wait(self: &mut WriteResult) -> FfiResult; + fn notify(self: &mut WriteResult, callback: UniquePtr) -> FfiResult; // UpsertWriter unsafe fn delete_upsert_writer(writer: *mut UpsertWriter); - fn upsert(self: &mut UpsertWriter, row: &GenericRowInner) -> FfiPtrResult; - fn delete_row(self: &mut UpsertWriter, row: &GenericRowInner) -> FfiPtrResult; + fn upsert(self: &mut UpsertWriter, row: &GenericRowInner, budget_ms: i64) -> FfiPtrResult; + fn delete_row( + self: &mut UpsertWriter, + row: &GenericRowInner, + budget_ms: i64, + ) -> FfiPtrResult; fn upsert_flush(self: &mut UpsertWriter) -> FfiResult; // Lookuper @@ -2065,6 +2087,12 @@ impl Table { self.has_pk } + /// The connection's configured write-buffer wait timeout (client.writer.buffer.wait-timeout), + /// shared by callback admission and buffer waits. UINT64_MAX means unbounded. + fn writer_buffer_wait_timeout_ms(&self) -> u64 { + self.connection.config().writer_buffer_wait_timeout_ms + } + fn create_upsert_writer(&self, column_indices: Vec) -> ffi::FfiPtrResult { let _enter = RUNTIME.enter(); @@ -2156,15 +2184,29 @@ unsafe fn delete_append_writer(writer: *mut AppendWriter) { } } +/// Convert a C++ submit budget (milliseconds) into an optional buffer-wait deadline. +/// Negative means "no caller budget": fall back to the writer's configured buffer +/// wait timeout. `>= 0` caps the wait (0 makes it non-blocking / fail fast). +fn budget_deadline(budget_ms: i64) -> Option { + if budget_ms < 0 { + None + } else { + Some(std::time::Instant::now() + std::time::Duration::from_millis(budget_ms as u64)) + } +} + impl AppendWriter { - fn append(&mut self, row: &GenericRowInner) -> ffi::FfiPtrResult { + fn append(&mut self, row: &GenericRowInner, budget_ms: i64) -> ffi::FfiPtrResult { let schema = self.table_info.get_schema(); let generic_row = match types::resolve_row_types(&row.row, Some(schema), 0) { Ok(r) => r, Err(e) => return client_err_ptr(e.to_string()), }; - let result_future = match self.inner.append(generic_row.as_ref()) { + let result_future = match self + .inner + .append_with_deadline(generic_row.as_ref(), budget_deadline(budget_ms)) + { Ok(f) => f, Err(e) => return err_ptr_from_core(&e), }; @@ -2175,7 +2217,12 @@ impl AppendWriter { ok_ptr(ptr as usize) } - fn append_arrow_batch(&mut self, array_ptr: usize, schema_ptr: usize) -> ffi::FfiPtrResult { + fn append_arrow_batch( + &mut self, + array_ptr: usize, + schema_ptr: usize, + budget_ms: i64, + ) -> ffi::FfiPtrResult { // Safety: C++ allocates these via `new ArrowArray/ArrowSchema` after a // successful `ExportRecordBatch`, so both pointers are valid heap // allocations that we take ownership of here. @@ -2194,7 +2241,10 @@ impl AppendWriter { let struct_array = arrow::array::StructArray::from(array_data); let batch = arrow::record_batch::RecordBatch::from(struct_array); - let result_future = match self.inner.append_arrow_batch(batch) { + let result_future = match self + .inner + .append_arrow_batch_with_deadline(batch, budget_deadline(budget_ms)) + { Ok(f) => f, Err(e) => return err_ptr_from_core(&e), }; @@ -2247,7 +2297,7 @@ unsafe fn delete_upsert_writer(writer: *mut UpsertWriter) { } impl UpsertWriter { - fn upsert(&mut self, row: &GenericRowInner) -> ffi::FfiPtrResult { + fn upsert(&mut self, row: &GenericRowInner, budget_ms: i64) -> ffi::FfiPtrResult { let schema = self.table_info.get_schema(); // Resolve types and pad to full schema width, so callers may set only // the fields they care about. @@ -2257,7 +2307,10 @@ impl UpsertWriter { Err(e) => return client_err_ptr(e.to_string()), }; - let result_future = match self.inner.upsert(generic_row.as_ref()) { + let result_future = match self + .inner + .upsert_with_deadline(generic_row.as_ref(), budget_deadline(budget_ms)) + { Ok(f) => f, Err(e) => return err_ptr_from_core(&e), }; @@ -2268,7 +2321,7 @@ impl UpsertWriter { ok_ptr(ptr as usize) } - fn delete_row(&mut self, row: &GenericRowInner) -> ffi::FfiPtrResult { + fn delete_row(&mut self, row: &GenericRowInner, budget_ms: i64) -> ffi::FfiPtrResult { let schema = self.table_info.get_schema(); // Resolve types and pad to full schema width, so callers may set only // the fields they care about. @@ -2278,7 +2331,10 @@ impl UpsertWriter { Err(e) => return client_err_ptr(e.to_string()), }; - let result_future = match self.inner.delete(generic_row.as_ref()) { + let result_future = match self + .inner + .delete_with_deadline(generic_row.as_ref(), budget_deadline(budget_ms)) + { Ok(f) => f, Err(e) => return err_ptr_from_core(&e), }; diff --git a/fluss-rust/bindings/cpp/src/table.cpp b/fluss-rust/bindings/cpp/src/table.cpp index 9601dcfecca..e65953b55ae 100644 --- a/fluss-rust/bindings/cpp/src/table.cpp +++ b/fluss-rust/bindings/cpp/src/table.cpp @@ -1281,14 +1281,23 @@ bool Table::HasPrimaryKey() const { TableAppend::TableAppend(ffi::Table* table) noexcept : table_(table) {} Result TableAppend::CreateWriter(AppendWriter& out) { + return CreateWriter(out, WriteCallbackOptions{}); +} + +Result TableAppend::CreateWriter(AppendWriter& out, const WriteCallbackOptions& options) { + if (options.max_pending_operations == 0) { + return utils::make_client_error("max_pending_operations must be greater than zero"); + } if (table_ == nullptr) { return utils::make_client_error("Table not available"); } + auto capacity = std::make_shared( + options.max_pending_operations, table_->writer_buffer_wait_timeout_ms()); auto ffi_result = table_->new_append_writer(); auto result = utils::from_ffi_result(ffi_result.result); if (result.Ok()) { - out = AppendWriter(utils::ptr_from_ffi(ffi_result)); + out = AppendWriter(utils::ptr_from_ffi(ffi_result), std::move(capacity)); } return result; } @@ -1339,11 +1348,20 @@ std::vector TableUpsert::ResolveNameProjection() const { } Result TableUpsert::CreateWriter(UpsertWriter& out) { + return CreateWriter(out, WriteCallbackOptions{}); +} + +Result TableUpsert::CreateWriter(UpsertWriter& out, const WriteCallbackOptions& options) { + if (options.max_pending_operations == 0) { + return utils::make_client_error("max_pending_operations must be greater than zero"); + } if (table_ == nullptr) { return utils::make_client_error("Table not available"); } try { + auto capacity = std::make_shared( + options.max_pending_operations, table_->writer_buffer_wait_timeout_ms()); auto resolved_indices = !column_names_.empty() ? ResolveNameProjection() : column_indices_; rust::Vec rust_indices; @@ -1353,7 +1371,8 @@ Result TableUpsert::CreateWriter(UpsertWriter& out) { auto ffi_result = table_->create_upsert_writer(std::move(rust_indices)); auto result = utils::from_ffi_result(ffi_result.result); if (result.Ok()) { - out = UpsertWriter(utils::ptr_from_ffi(ffi_result)); + out = UpsertWriter(utils::ptr_from_ffi(ffi_result), + std::move(capacity)); } return result; } catch (const std::exception& e) { @@ -1605,13 +1624,19 @@ Result WriteResult::Wait() { return utils::from_ffi_result(ffi_result); } +Result WriteResult::Notify(std::unique_ptr callback) { + return utils::from_ffi_result(inner_->notify(std::move(callback))); +} + // ============================================================================ // AppendWriter // ============================================================================ AppendWriter::AppendWriter() noexcept = default; -AppendWriter::AppendWriter(ffi::AppendWriter* writer) noexcept : writer_(writer) {} +AppendWriter::AppendWriter(ffi::AppendWriter* writer, + std::shared_ptr callback_capacity) noexcept + : writer_(writer), callback_capacity_(std::move(callback_capacity)) {} AppendWriter::~AppendWriter() noexcept { Destroy(); } @@ -1622,7 +1647,8 @@ void AppendWriter::Destroy() noexcept { } } -AppendWriter::AppendWriter(AppendWriter&& other) noexcept : writer_(other.writer_) { +AppendWriter::AppendWriter(AppendWriter&& other) noexcept + : writer_(other.writer_), callback_capacity_(std::move(other.callback_capacity_)) { other.writer_ = nullptr; } @@ -1630,6 +1656,7 @@ AppendWriter& AppendWriter::operator=(AppendWriter&& other) noexcept { if (this != &other) { Destroy(); writer_ = other.writer_; + callback_capacity_ = std::move(other.callback_capacity_); other.writer_ = nullptr; } return *this; @@ -1643,6 +1670,11 @@ Result AppendWriter::Append(const GenericRow& row) { } Result AppendWriter::Append(const GenericRow& row, WriteResult& out) { + return AppendWithBudget(row, out, -1); +} + +Result AppendWriter::AppendWithBudget(const GenericRow& row, WriteResult& out, + int64_t submit_budget_ms) { if (!Available()) { return utils::make_client_error("AppendWriter not available"); } @@ -1650,7 +1682,7 @@ Result AppendWriter::Append(const GenericRow& row, WriteResult& out) { return utils::make_client_error("GenericRow not available"); } - auto ffi_result = writer_->append(*row.inner_); + auto ffi_result = writer_->append(*row.inner_, submit_budget_ms); auto result = utils::from_ffi_result(ffi_result.result); if (result.Ok()) { out = WriteResult(utils::ptr_from_ffi(ffi_result)); @@ -1658,6 +1690,31 @@ Result AppendWriter::Append(const GenericRow& row, WriteResult& out) { return result; } +Result AppendWriter::Append(const GenericRow& row, WriteCallback callback) { + if (!callback) { + return utils::make_client_error("Write callback must not be empty"); + } + auto executor_status = utils::from_ffi_result(ffi::ensure_callback_executor()); + if (!executor_status.Ok()) { + return executor_status; + } + // Allocate before submission so an allocation failure cannot lose an + // already accepted write's completion notification. + auto completion = std::make_unique(std::move(callback)); + // Share the capacity and buffer wait budget from client.writer.buffer.wait-timeout. + const auto submit_start = std::chrono::steady_clock::now(); + auto reserved = completion->Reserve(callback_capacity_); + if (!reserved.Ok()) { + return reserved; + } + WriteResult pending; + auto result = AppendWithBudget(row, pending, callback_capacity_->RemainingBudgetMs(submit_start)); + if (result.Ok()) { + return pending.Notify(std::move(completion)); + } + return result; +} + Result AppendWriter::AppendArrowBatch(const std::shared_ptr& batch) { WriteResult wr; return AppendArrowBatch(batch, wr); @@ -1665,6 +1722,11 @@ Result AppendWriter::AppendArrowBatch(const std::shared_ptr& Result AppendWriter::AppendArrowBatch(const std::shared_ptr& batch, WriteResult& out) { + return AppendArrowBatchWithBudget(batch, out, -1); +} + +Result AppendWriter::AppendArrowBatchWithBudget(const std::shared_ptr& batch, + WriteResult& out, int64_t submit_budget_ms) { if (!Available()) { return utils::make_client_error("AppendWriter not available"); } @@ -1687,7 +1749,8 @@ Result AppendWriter::AppendArrowBatch(const std::shared_ptr& // Rust takes ownership of both pointers immediately via Box::from_raw(), // so after this call C++ must NOT free them. auto ffi_result = writer_->append_arrow_batch(reinterpret_cast(array_heap), - reinterpret_cast(schema_heap)); + reinterpret_cast(schema_heap), + submit_budget_ms); auto result = utils::from_ffi_result(ffi_result.result); if (result.Ok()) { out.Destroy(); @@ -1696,13 +1759,44 @@ Result AppendWriter::AppendArrowBatch(const std::shared_ptr& return result; } +Result AppendWriter::AppendArrowBatch(const std::shared_ptr& batch, + WriteCallback callback) { + if (!callback) { + return utils::make_client_error("Write callback must not be empty"); + } + auto executor_status = utils::from_ffi_result(ffi::ensure_callback_executor()); + if (!executor_status.Ok()) { + return executor_status; + } + auto completion = std::make_unique(std::move(callback)); + const auto submit_start = std::chrono::steady_clock::now(); + auto reserved = completion->Reserve(callback_capacity_); + if (!reserved.Ok()) { + return reserved; + } + WriteResult pending; + auto result = + AppendArrowBatchWithBudget(batch, pending, callback_capacity_->RemainingBudgetMs(submit_start)); + if (result.Ok()) { + return pending.Notify(std::move(completion)); + } + return result; +} + Result AppendWriter::Flush() { + if (ffi::WriteCallbackCapacity::InCallback()) { + return utils::make_client_error("Flush cannot be called from a write callback"); + } if (!Available()) { return utils::make_client_error("AppendWriter not available"); } auto ffi_result = writer_->flush(); - return utils::from_ffi_result(ffi_result); + auto result = utils::from_ffi_result(ffi_result); + if (!result.Ok()) return result; + // Writes are flushed; block until their callbacks drain so Flush is a real barrier. + callback_capacity_->AwaitAll(); + return {}; } // ============================================================================ @@ -1711,7 +1805,9 @@ Result AppendWriter::Flush() { UpsertWriter::UpsertWriter() noexcept = default; -UpsertWriter::UpsertWriter(ffi::UpsertWriter* writer) noexcept : writer_(writer) {} +UpsertWriter::UpsertWriter(ffi::UpsertWriter* writer, + std::shared_ptr callback_capacity) noexcept + : writer_(writer), callback_capacity_(std::move(callback_capacity)) {} UpsertWriter::~UpsertWriter() noexcept { Destroy(); } @@ -1722,7 +1818,8 @@ void UpsertWriter::Destroy() noexcept { } } -UpsertWriter::UpsertWriter(UpsertWriter&& other) noexcept : writer_(other.writer_) { +UpsertWriter::UpsertWriter(UpsertWriter&& other) noexcept + : writer_(other.writer_), callback_capacity_(std::move(other.callback_capacity_)) { other.writer_ = nullptr; } @@ -1730,6 +1827,7 @@ UpsertWriter& UpsertWriter::operator=(UpsertWriter&& other) noexcept { if (this != &other) { Destroy(); writer_ = other.writer_; + callback_capacity_ = std::move(other.callback_capacity_); other.writer_ = nullptr; } return *this; @@ -1743,6 +1841,11 @@ Result UpsertWriter::Upsert(const GenericRow& row) { } Result UpsertWriter::Upsert(const GenericRow& row, WriteResult& out) { + return UpsertWithBudget(row, out, -1); +} + +Result UpsertWriter::UpsertWithBudget(const GenericRow& row, WriteResult& out, + int64_t submit_budget_ms) { if (!Available()) { return utils::make_client_error("UpsertWriter not available"); } @@ -1750,7 +1853,7 @@ Result UpsertWriter::Upsert(const GenericRow& row, WriteResult& out) { return utils::make_client_error("GenericRow not available"); } - auto ffi_result = writer_->upsert(*row.inner_); + auto ffi_result = writer_->upsert(*row.inner_, submit_budget_ms); auto result = utils::from_ffi_result(ffi_result.result); if (result.Ok()) { out = WriteResult(utils::ptr_from_ffi(ffi_result)); @@ -1758,12 +1861,39 @@ Result UpsertWriter::Upsert(const GenericRow& row, WriteResult& out) { return result; } +Result UpsertWriter::Upsert(const GenericRow& row, WriteCallback callback) { + if (!callback) { + return utils::make_client_error("Write callback must not be empty"); + } + auto executor_status = utils::from_ffi_result(ffi::ensure_callback_executor()); + if (!executor_status.Ok()) { + return executor_status; + } + auto completion = std::make_unique(std::move(callback)); + const auto submit_start = std::chrono::steady_clock::now(); + auto reserved = completion->Reserve(callback_capacity_); + if (!reserved.Ok()) { + return reserved; + } + WriteResult pending; + auto result = UpsertWithBudget(row, pending, callback_capacity_->RemainingBudgetMs(submit_start)); + if (result.Ok()) { + return pending.Notify(std::move(completion)); + } + return result; +} + Result UpsertWriter::Delete(const GenericRow& row) { WriteResult wr; return Delete(row, wr); } Result UpsertWriter::Delete(const GenericRow& row, WriteResult& out) { + return DeleteWithBudget(row, out, -1); +} + +Result UpsertWriter::DeleteWithBudget(const GenericRow& row, WriteResult& out, + int64_t submit_budget_ms) { if (!Available()) { return utils::make_client_error("UpsertWriter not available"); } @@ -1771,7 +1901,7 @@ Result UpsertWriter::Delete(const GenericRow& row, WriteResult& out) { return utils::make_client_error("GenericRow not available"); } - auto ffi_result = writer_->delete_row(*row.inner_); + auto ffi_result = writer_->delete_row(*row.inner_, submit_budget_ms); auto result = utils::from_ffi_result(ffi_result.result); if (result.Ok()) { out = WriteResult(utils::ptr_from_ffi(ffi_result)); @@ -1779,13 +1909,42 @@ Result UpsertWriter::Delete(const GenericRow& row, WriteResult& out) { return result; } +Result UpsertWriter::Delete(const GenericRow& row, WriteCallback callback) { + if (!callback) { + return utils::make_client_error("Write callback must not be empty"); + } + auto executor_status = utils::from_ffi_result(ffi::ensure_callback_executor()); + if (!executor_status.Ok()) { + return executor_status; + } + auto completion = std::make_unique(std::move(callback)); + const auto submit_start = std::chrono::steady_clock::now(); + auto reserved = completion->Reserve(callback_capacity_); + if (!reserved.Ok()) { + return reserved; + } + WriteResult pending; + auto result = DeleteWithBudget(row, pending, callback_capacity_->RemainingBudgetMs(submit_start)); + if (result.Ok()) { + return pending.Notify(std::move(completion)); + } + return result; +} + Result UpsertWriter::Flush() { + if (ffi::WriteCallbackCapacity::InCallback()) { + return utils::make_client_error("Flush cannot be called from a write callback"); + } if (!Available()) { return utils::make_client_error("UpsertWriter not available"); } auto ffi_result = writer_->upsert_flush(); - return utils::from_ffi_result(ffi_result); + auto result = utils::from_ffi_result(ffi_result); + if (!result.Ok()) return result; + // Writes are flushed; block until their callbacks drain so Flush is a real barrier. + callback_capacity_->AwaitAll(); + return {}; } // ============================================================================ diff --git a/fluss-rust/bindings/cpp/src/write_callback.hpp b/fluss-rust/bindings/cpp/src/write_callback.hpp new file mode 100644 index 00000000000..7eae1b7820e --- /dev/null +++ b/fluss-rust/bindings/cpp/src/write_callback.hpp @@ -0,0 +1,177 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "fluss.hpp" +#include "rust/cxx.h" + +namespace fluss { +namespace ffi { + +/// Per-writer admission control; independent of Rust buffer memory and ACK completion. +class WriteCallbackCapacity { + public: + /// `wait_timeout_ms` is the connection's client.writer.buffer.wait-timeout, used as + /// the shared budget for capacity and buffer waits. UINT64_MAX means block until a slot frees. + WriteCallbackCapacity(size_t max_pending_operations, uint64_t wait_timeout_ms) + : max_pending_(max_pending_operations), wait_timeout_ms_(wait_timeout_ms) {} + + Result Acquire() { + std::unique_lock lock(mutex_); + if (pending_ == max_pending_) { + auto has_slot = [&] { return pending_ < max_pending_; }; + // Fail fast from within a callback to avoid stalling the shared workers on + // their own capacity; a zero budget also rejects immediately. + if (in_callback_ || wait_timeout_ms_ == 0) { + return {ErrorCode::CLIENT_ERROR, "Write callback capacity is full"}; + } + if (IsUnbounded()) { + available_.wait(lock, has_slot); + } else if (!available_.wait_for(lock, std::chrono::milliseconds(wait_timeout_ms_), + has_slot)) { + return {ErrorCode::CLIENT_ERROR, "Timed out waiting for write callback capacity"}; + } + } + ++pending_; + return {}; + } + + void Release() noexcept { + { + std::lock_guard lock(mutex_); + --pending_; + } + // notify_all: Acquire() waiters and the AwaitAll() waiter share this condvar, + // so waking only one risks waking AwaitAll() (still pending) while an Acquire() + // waiter keeps sleeping despite the freed slot. + available_.notify_all(); + } + + /// True only while this thread executes a user callback or destroys its captures. + static bool InCallback() { return in_callback_; } + + /// Wait until every reserved callback and its captures have finished. + /// Flush rejects callback reentry before starting any write flush. + void AwaitAll() { + std::unique_lock lock(mutex_); + available_.wait(lock, [&] { return pending_ == 0; }); + } + + /// Milliseconds left in the client.writer.buffer.wait-timeout budget since `start`, so + /// the buffer-backpressure wait plus the capacity reservation stay within one timeout + /// (Kafka max.block.ms style). Floored at 0 (0 = fail fast). Returns -1 when the timeout + /// is unbounded, letting the buffer wait fall back to the writer's configured timeout. + int64_t RemainingBudgetMs(std::chrono::steady_clock::time_point start) const { + if (IsUnbounded()) { + return -1; + } + auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start); + int64_t remaining = static_cast(wait_timeout_ms_) - elapsed.count(); + return remaining > 0 ? remaining : 0; + } + + private: + friend class WriteCallback; + inline static thread_local bool in_callback_ = false; + bool IsUnbounded() const { return wait_timeout_ms_ == std::numeric_limits::max(); } + const size_t max_pending_; + const uint64_t wait_timeout_ms_; + std::mutex mutex_; + std::condition_variable available_; + size_t pending_ = 0; +}; + +/// Owns a callback transferred to Rust. Access is exclusive, never concurrent. +class WriteCallback { + public: + explicit WriteCallback(fluss::WriteCallback callback) : callback_(std::move(callback)) {} + + WriteCallback(const WriteCallback&) = delete; + WriteCallback& operator=(const WriteCallback&) = delete; + + /// Reserve before entering Rust. Destruction also returns capacity on submission failure. + Result Reserve(std::shared_ptr capacity) { + if (!capacity) { + return {ErrorCode::CLIENT_ERROR, "Writer not available"}; + } + auto result = capacity->Acquire(); + if (result.Ok()) { + reservation_.capacity = std::move(capacity); + } + return result; + } + + /// Invoke once, containing all C++ exceptions on this side of the FFI boundary. + void Complete(int32_t error_code, rust::Str error_message) noexcept { + // Release captures before the reservation, even if this wrapper outlives Complete(). + Reservation reservation{std::move(reservation_.capacity)}; + CallbackScope scope; + // Moving std::function alone need not empty the source. Swap with an + // empty function so captures are released even if the callback throws. + fluss::WriteCallback callback; + callback.swap(callback_); + Result result; + result.error_code = error_code; + try { + result.error_message = std::string(error_message); + } catch (...) { + // Error text is best-effort; allocation failure must not skip completion. + std::fprintf(stderr, "Fluss write callback could not copy error text (code %d)\n", + error_code); + } + try { + callback(WriteCompletion{std::move(result)}); + } catch (const std::exception& e) { + std::fprintf(stderr, "Fluss write callback threw an exception: %s\n", e.what()); + } catch (...) { + std::fprintf(stderr, "Fluss write callback threw an unknown exception\n"); + } + } + + private: + struct Reservation { + std::shared_ptr capacity; + ~Reservation() { + if (capacity) { + capacity->Release(); + } + } + }; + + struct CallbackScope { + bool previous = std::exchange(WriteCallbackCapacity::in_callback_, true); + ~CallbackScope() { WriteCallbackCapacity::in_callback_ = previous; } + }; + + // Member order keeps captures alive until invocation, but not past capacity release. + Reservation reservation_; + fluss::WriteCallback callback_; +}; + +} // namespace ffi +} // namespace fluss diff --git a/fluss-rust/bindings/cpp/src/write_callback.rs b/fluss-rust/bindings/cpp/src/write_callback.rs new file mode 100644 index 00000000000..6810a8ef013 --- /dev/null +++ b/fluss-rust/bindings/cpp/src/write_callback.rs @@ -0,0 +1,427 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#[cfg(test)] +use std::future::Future; +use std::panic::{AssertUnwindSafe, catch_unwind}; +use std::sync::{LazyLock, mpsc}; +use std::thread::{self, JoinHandle}; + +use crate::{RUNTIME, WriteResult, client_err, err_from_core_error, ffi, ok_result}; + +type Completion = Box; + +// One process-wide worker preserves dispatch order without running user code +// on I/O threads. Admission is bounded independently for each C++ writer. +// Initialize before accepting any callback write. Failure is sticky and returned +// synchronously; silently switching to a parallel pool would break ordering. +static CALLBACK_EXECUTOR: LazyLock> = + LazyLock::new(CallbackExecutor::new); + +pub(crate) fn ensure_callback_executor() -> ffi::FfiResult { + executor_status(&CALLBACK_EXECUTOR) +} + +fn executor_status(executor: &std::io::Result) -> ffi::FfiResult { + match executor { + Ok(_) => ok_result(), + Err(error) => client_err(format!( + "Cannot initialize write callback executor: {error}" + )), + } +} + +struct CallbackExecutor { + sender: Option>, + worker: Option>, +} + +impl CallbackExecutor { + fn new() -> std::io::Result { + let (sender, receiver) = mpsc::channel::(); + let worker = thread::Builder::new() + .name("fluss-callback".to_string()) + .spawn(move || { + while let Ok(completion) = receiver.recv() { + // This is a dedicated OS thread, not a Tokio runtime worker. + // No receiver mutex or user-configurable worker count is needed. + if catch_unwind(AssertUnwindSafe(completion)).is_err() { + eprintln!("Fluss callback worker contained a Rust panic"); + } + } + })?; + Ok(Self { + sender: Some(sender), + worker: Some(worker), + }) + } + + fn enqueue(&self, completion: Completion) -> Result<(), Completion> { + // An unbounded completion queue keeps slow user callbacks from blocking + // async I/O workers. Per-writer admission bounds callback operations, + // not the bytes retained by captures or total process memory. + self.sender + .as_ref() + .unwrap() + .send(completion) + .map_err(|error| error.0) + } +} + +impl Drop for CallbackExecutor { + fn drop(&mut self) { + // Tests own executors and verify drain/release. The process-wide static + // lives until process exit and is not automatically drained there. + drop(self.sender.take()); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } + } +} + +// SAFETY: The C++ wrapper is transferred by UniquePtr and accessed exclusively +// by one batch registration, then one callback worker. It is never shared concurrently. +// The public C++ contract requires captures to support background execution. +unsafe impl Send for ffi::WriteCallback {} + +impl WriteResult { + pub(crate) fn notify( + &mut self, + mut callback: cxx::UniquePtr, + ) -> ffi::FfiResult { + if callback.is_null() { + return client_err("Write callback must not be empty".to_string()); + } + let Some(future) = self.inner.take() else { + return client_err("WriteResult already consumed".to_string()); + }; + dispatch_write(future, move |result| { + callback + .pin_mut() + .complete(result.error_code, &result.error_message); + }); + ok_result() + } +} + +fn dispatch_write( + future: fluss::client::WriteResultFuture, + callback: impl FnOnce(ffi::FfiResult) + Send + 'static, +) { + // Force worker initialization before registering with an in-flight batch. + assert_eq!(ensure_callback_executor().error_code, 0); + let callback = move |result| callback(to_ffi_result(result)); + if let Err((future, callback)) = future.try_on_complete(callback, dispatch_batch) { + // Only futures already polled before registration need this path. + // Normal C++ Append/Upsert/Delete never poll before registering. + RUNTIME.spawn(async move { + let result = future.await; + deliver( + CALLBACK_EXECUTOR + .as_ref() + .expect("callback executor initialized before submission"), + Box::new(move || callback(result)), + ); + }); + } +} + +fn dispatch_batch(batch: fluss::client::WriteCallbackBatch) { + // Deliver the whole batch as one job so its callbacks stay together and the + // single worker runs them in completion order. + deliver( + CALLBACK_EXECUTOR + .as_ref() + .expect("callback executor initialized before submission"), + Box::new(move || batch.run()), + ); +} + +fn to_ffi_result(result: Result<(), fluss::error::Error>) -> ffi::FfiResult { + match result { + Ok(()) => ok_result(), + Err(e) => err_from_core_error(&e), + } +} + +#[cfg(test)] +fn dispatch( + future: impl Future> + Send + 'static, + callback: impl FnOnce(ffi::FfiResult) + Send + 'static, +) { + let executor = CALLBACK_EXECUTOR + .as_ref() + .expect("callback executor initialized"); + RUNTIME.spawn(async move { + let result = to_ffi_result(future.await); + deliver(executor, Box::new(move || callback(result))); + }); +} + +fn deliver(executor: &CallbackExecutor, completion: Completion) { + if executor.enqueue(completion).is_err() { + // The static executor is never shut down and contains callback panics. + // Disconnection is an internal invariant violation, not an overload policy. + // Do not silently lose accepted notifications or run them out of order. + eprintln!("Fluss callback executor unexpectedly disconnected"); + std::process::abort(); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{Arc, mpsc}; + use std::thread; + use std::time::Duration; + + use super::{CallbackExecutor, dispatch, dispatch_write, executor_status}; + use crate::{CLIENT_ERROR_CODE, RUNTIME}; + + #[test] + fn test_direct_batch_completion_runs_off_runtime_and_releases_capture() { + let (tx, rx) = mpsc::channel(); + let capture = Arc::new(()); + let weak = Arc::downgrade(&capture); + RUNTIME.block_on(async { + dispatch_write( + fluss::client::WriteResultFuture::join(Vec::new()), + move |r| { + // Empty/previously completed batches must still use the executor. + let answer = RUNTIME.block_on(async { 42 }); + tx.send((r.error_code, answer, capture)).unwrap(); + }, + ); + }); + let (code, answer, capture) = rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!((code, answer), (0, 42)); + drop(capture); + assert!(weak.upgrade().is_none()); + assert!(matches!( + rx.try_recv(), + Err(mpsc::TryRecvError::Disconnected) + )); + } + + #[test] + fn test_callback_waits_asynchronously_for_acknowledgment() { + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + let (callback_tx, callback_rx) = mpsc::channel(); + dispatch( + async move { + ack_rx.await.unwrap(); + Ok(()) + }, + move |result| callback_tx.send(result).unwrap(), + ); + assert!(matches!( + callback_rx.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + ack_tx.send(()).unwrap(); + let result = callback_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(result.error_code, 0); + assert!(result.error_message.is_empty()); + assert!(matches!( + callback_rx.recv_timeout(Duration::from_secs(10)), + Err(mpsc::RecvTimeoutError::Disconnected) + )); + } + + #[test] + fn test_callback_preserves_server_error() { + let (tx, rx) = mpsc::channel(); + dispatch( + async { + Err(fluss::error::Error::FlussAPIError { + api_error: fluss::rpc::ApiError { + code: 57, + message: "Deletion is disabled".to_string(), + }, + }) + }, + move |result| tx.send(result).unwrap(), + ); + let result = rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(result.error_code, 57); + assert_eq!(result.error_message, "Deletion is disabled"); + } + + #[test] + fn test_callback_preserves_client_error() { + let (tx, rx) = mpsc::channel(); + dispatch( + async { + Err(fluss::error::Error::UnexpectedError { + message: "Writer closed".to_string(), + source: None, + }) + }, + move |result| tx.send(result).unwrap(), + ); + let result = rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(result.error_code, CLIENT_ERROR_CODE); + assert!(result.error_message.contains("Writer closed")); + } + + #[test] + fn test_callback_can_reenter_synchronous_runtime_calls() { + let (tx, rx) = mpsc::channel(); + dispatch(async { Ok(()) }, move |_| { + // block_on would panic if the callback ran on an async worker. + let result = RUNTIME.block_on(async { RUNTIME.spawn(async { 42 }).await.unwrap() }); + tx.send(result).unwrap(); + }); + assert_eq!(rx.recv_timeout(Duration::from_secs(10)).unwrap(), 42); + } + + #[test] + fn test_slow_callback_does_not_block_runtime_or_run_callbacks_concurrently() { + let executor = CallbackExecutor::new().unwrap(); + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let (done_tx, done_rx) = mpsc::channel(); + assert!( + executor + .enqueue(Box::new(move || { + started_tx.send(thread::current().id()).unwrap(); + release_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + })) + .is_ok() + ); + let worker_id = started_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_ne!(worker_id, thread::current().id()); + for _ in 0..256 { + let done_tx = done_tx.clone(); + assert!( + executor + .enqueue(Box::new(move || { + done_tx.send(thread::current().id()).unwrap(); + })) + .is_ok() + ); + } + assert!(matches!(done_rx.try_recv(), Err(mpsc::TryRecvError::Empty))); + assert_eq!( + RUNTIME.block_on(async { + tokio::time::timeout(Duration::from_secs(5), RUNTIME.spawn(async { 42 })) + .await + .unwrap() + .unwrap() + }), + 42 + ); + release_tx.send(()).unwrap(); + drop(executor); + for _ in 0..256 { + assert_eq!( + done_rx.recv_timeout(Duration::from_secs(10)).unwrap(), + worker_id + ); + } + } + + #[test] + fn test_executor_drains_and_survives_a_panicking_callback() { + let executor = CallbackExecutor::new().unwrap(); + let completed = Arc::new(AtomicUsize::new(0)); + assert!( + executor + .enqueue(Box::new(|| panic!("test callback panic"))) + .is_ok() + ); + for _ in 0..1000 { + let completed = Arc::clone(&completed); + assert!( + executor + .enqueue(Box::new(move || { + completed.fetch_add(1, Ordering::Relaxed); + })) + .is_ok() + ); + } + drop(executor); + assert_eq!(completed.load(Ordering::Relaxed), 1000); + assert_eq!(Arc::strong_count(&completed), 1); + } + + #[test] + fn test_single_worker_runs_callbacks_in_enqueue_order() { + // A single worker must not reorder callbacks. This is the executor-level + // guarantee behind same-bucket completions reporting in the order they + // finished: the results come out exactly as enqueued, not just once each. + let executor = CallbackExecutor::new().unwrap(); + let (tx, rx) = mpsc::channel(); + for index in 0..1000 { + let tx = tx.clone(); + assert!( + executor + .enqueue(Box::new(move || { + tx.send(index).unwrap(); + })) + .is_ok() + ); + } + drop(tx); + let observed: Vec<_> = (0..1000) + .map(|_| rx.recv_timeout(Duration::from_secs(10)).unwrap()) + .collect(); + assert_eq!(observed, (0..1000).collect::>()); + } + + #[test] + fn test_concurrent_producers_complete_each_job_once() { + let executor = Arc::new(CallbackExecutor::new().unwrap()); + let (tx, rx) = mpsc::channel(); + let mut producers = Vec::new(); + for producer in 0..4 { + let executor = Arc::clone(&executor); + let tx = tx.clone(); + producers.push(thread::spawn(move || { + for index in 0..1000 { + let tx = tx.clone(); + assert!( + executor + .enqueue(Box::new(move || { + tx.send(producer * 1000 + index).unwrap(); + })) + .is_ok() + ); + } + })); + } + for producer in producers { + producer.join().unwrap(); + } + drop(executor); + drop(tx); + let mut completed: Vec<_> = rx.into_iter().collect(); + completed.sort_unstable(); + assert_eq!(completed, (0..4000).collect::>()); + } + + #[test] + fn test_executor_initialization_error_is_reported_without_fallback() { + let unavailable = Err(std::io::Error::other("thread creation failed")); + let status = executor_status(&unavailable); + assert_eq!(status.error_code, CLIENT_ERROR_CODE); + assert!(status.error_message.contains("thread creation failed")); + assert!(unavailable.is_err()); + let ready = Ok(CallbackExecutor::new().unwrap()); + assert_eq!(executor_status(&ready).error_code, 0); + } +} diff --git a/fluss-rust/bindings/cpp/test/test_write_callback.cpp b/fluss-rust/bindings/cpp/test/test_write_callback.cpp new file mode 100644 index 00000000000..308880d4431 --- /dev/null +++ b/fluss-rust/bindings/cpp/test/test_write_callback.cpp @@ -0,0 +1,692 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include + +#include +#include +#include +#include +#include + +#include "test_utils.h" +#include "write_callback.hpp" + +namespace { + +class Completion { + public: + void Reset() { + std::lock_guard lock(mutex_); + results_.clear(); + thread_ = std::thread::id{}; + } + + void Record(fluss::Result result) { + std::lock_guard lock(mutex_); + results_.push_back(std::move(result)); + thread_ = std::this_thread::get_id(); + ready_.notify_all(); + } + + bool Await(size_t count = 1) { + std::unique_lock lock(mutex_); + return ready_.wait_for(lock, std::chrono::seconds(10), + [&] { return results_.size() >= count; }); + } + + std::vector Results() { + std::lock_guard lock(mutex_); + return results_; + } + + std::thread::id Thread() { + std::lock_guard lock(mutex_); + return thread_; + } + + private: + std::mutex mutex_; + std::condition_variable ready_; + std::vector results_; + std::thread::id thread_; +}; + +// A plain function pointer has no capture. This state lives for the process. +Completion function_completion; + +void RecordFunctionCallback(const fluss::WriteCompletion& completion) { + function_completion.Record(completion.result); +} + +struct Lifetime { + std::promise released; + ~Lifetime() { released.set_value(); } +}; + +} // namespace + +class WriteCallbackTest : public ::testing::Test { + protected: + void CreateTable(bool primary_key = false, bool disable_delete = false) { + auto& env = *fluss_test::FlussTestEnvironment::Instance(); + auto builder = fluss::Schema::NewBuilder() + .AddColumn("id", fluss::DataType::Int()) + .AddColumn("value", fluss::DataType::String()); + if (primary_key) { + builder.SetPrimaryKeys({"id"}); + } + auto descriptor_builder = fluss::TableDescriptor::NewBuilder() + .SetSchema(builder.Build()) + .SetBucketCount(3) + .SetBucketKeys({"id"}) + .SetProperty("table.replication.factor", "1"); + if (disable_delete) { + descriptor_builder.SetProperty("table.delete.behavior", "disable"); + } + auto descriptor = descriptor_builder.Build(); + table_path_ = fluss::TablePath( + "fluss", std::string("cpp_callback_") + + ::testing::UnitTest::GetInstance()->current_test_info()->name()); + fluss_test::CreateTable(env.GetAdmin(), table_path_, descriptor); + auto result = env.GetConnection().GetTable(table_path_, table_); + ASSERT_OK(result); + } + + fluss::GenericRow Row(int32_t id = 1) { + fluss::GenericRow row(2); + row.SetInt32(0, id); + row.SetString(1, "callback"); + return row; + } + + fluss::TablePath table_path_; + fluss::Table table_; +}; + +TEST_F(WriteCallbackTest, AppendAcceptsFunctionPointer) { + function_completion.Reset(); + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + + auto submitted = writer.Append(Row(), &RecordFunctionCallback); + ASSERT_OK(submitted); + ASSERT_TRUE(function_completion.Await()); + auto results = function_completion.Results(); + ASSERT_EQ(results.size(), 1u); + EXPECT_OK(results.front()); + EXPECT_NE(function_completion.Thread(), std::this_thread::get_id()); + + // The old acknowledgment and fire-and-forget overloads still work. + fluss::WriteResult pending; + ASSERT_OK(writer.Append(Row(2), pending)); + ASSERT_OK(pending.Wait()); + ASSERT_OK(writer.Append(Row(3))); + ASSERT_OK(writer.Flush()); +} + +TEST_F(WriteCallbackTest, FlushWaitsForPendingCallbacks) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto started = std::make_shared(); + auto finished = std::make_shared(); + auto gate = std::make_shared>(); + auto resume = gate->get_future().share(); + auto lifetime = std::make_shared(); + auto released = lifetime->released.get_future(); + std::weak_ptr weak_lifetime = lifetime; + { + auto row = Row(); + fluss::WriteCallback callback = [started, finished, resume, owned = std::move(lifetime)]( + const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + started->Record(result); + // Bounded even when a preceding assertion fails. + resume.wait_for(std::chrono::seconds(20)); + finished->Record(result); + }; + ASSERT_OK(writer.Append(row, std::move(callback))); + } + ASSERT_TRUE(started->Await()); + EXPECT_FALSE(weak_lifetime.expired()); + // Flush now waits for pending callbacks, not just for server ACK. + // The callback is blocked on the gate, so Flush must not return yet. + std::promise flush_started; + auto flush = std::async(std::launch::async, [&] { + flush_started.set_value(); + return writer.Flush(); + }); + flush_started.get_future().wait(); + EXPECT_EQ(flush.wait_for(std::chrono::milliseconds(25)), std::future_status::timeout); + gate->set_value(); + ASSERT_OK(flush.get()); + // After Flush returns, the callback has finished and captures are released. + EXPECT_FALSE(finished->Results().empty()); + EXPECT_EQ(released.wait_for(std::chrono::seconds(10)), std::future_status::ready); + EXPECT_TRUE(weak_lifetime.expired()); + EXPECT_OK(finished->Results().front()); +} + +TEST_F(WriteCallbackTest, AppendArrowBatchNotifiesOnce) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto completion = std::make_shared(); + { + arrow::Int32Builder ids; + arrow::StringBuilder values; + ASSERT_TRUE(ids.AppendValues({1, 2, 3, 4, 5, 6}).ok()); + ASSERT_TRUE(values.AppendValues({"a", "b", "c", "d", "e", "f"}).ok()); + auto batch = + arrow::RecordBatch::Make(arrow::schema({arrow::field("id", arrow::int32()), + arrow::field("value", arrow::utf8())}), + 6, {ids.Finish().ValueOrDie(), values.Finish().ValueOrDie()}); + ASSERT_OK(writer.AppendArrowBatch(batch, + [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + })); + } + ASSERT_TRUE(completion->Await()); + ASSERT_EQ(completion->Results().size(), 1u); + EXPECT_OK(completion->Results().front()); + ASSERT_OK(writer.Flush()); +} + +TEST_F(WriteCallbackTest, UpsertAndDeleteNotifyCompletion) { + CreateTable(true); + fluss::UpsertWriter writer; + ASSERT_OK(table_.NewUpsert().CreateWriter(writer)); + auto completion = std::make_shared(); + fluss::WriteCallback callback = [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + }; + ASSERT_OK(writer.Upsert(Row(), callback)); + ASSERT_TRUE(completion->Await()); + + fluss::Lookuper lookuper; + ASSERT_OK(table_.NewLookup().CreateLookuper(lookuper)); + fluss::GenericRow key(2); + key.SetInt32(0, 1); + fluss::LookupResult found; + ASSERT_OK(lookuper.Lookup(key, found)); + ASSERT_TRUE(found.Found()); + + ASSERT_OK(writer.Delete(key, callback)); + ASSERT_TRUE(completion->Await(2)); + auto results = completion->Results(); + ASSERT_EQ(results.size(), 2u); + EXPECT_OK(results[0]); + EXPECT_OK(results[1]); + fluss::LookupResult deleted; + ASSERT_OK(lookuper.Lookup(key, deleted)); + EXPECT_FALSE(deleted.Found()); +} + +TEST_F(WriteCallbackTest, ServerRejectionIsReportedThroughCallback) { + CreateTable(true, true); + fluss::UpsertWriter writer; + ASSERT_OK(table_.NewUpsert().CreateWriter(writer)); + fluss::WriteResult pending; + ASSERT_OK(writer.Upsert(Row(), pending)); + ASSERT_OK(pending.Wait()); + + auto completion = std::make_shared(); + auto submitted = writer.Delete(Row(), [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + }); + // The write is accepted locally; only the callback reports server rejection. + ASSERT_OK(submitted); + ASSERT_TRUE(completion->Await()); + auto results = completion->Results(); + ASSERT_EQ(results.size(), 1u); + EXPECT_EQ(results.front().error_code, fluss::ErrorCode::DELETION_DISABLED_EXCEPTION); + EXPECT_NE(results.front().error_message.find("disabled"), std::string::npos); +} + +TEST_F(WriteCallbackTest, MultipleOutstandingWritesEachNotifyOnce) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto completion = std::make_shared(); + for (int32_t id = 0; id < 64; ++id) { + ASSERT_OK(writer.Append(Row(id), [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + })); + } + ASSERT_TRUE(completion->Await(64)); + ASSERT_OK(writer.Flush()); + auto results = completion->Results(); + ASSERT_EQ(results.size(), 64u); + for (const auto& result : results) { + EXPECT_OK(result); + } +} + +TEST_F(WriteCallbackTest, RejectedSubmissionDoesNotInvokeCallback) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto completion = std::make_shared(); + auto lifetime = std::make_shared(); + auto released = lifetime->released.get_future(); + fluss::GenericRow invalid(1); + invalid.SetInt32(0, 1); // Table requires two columns. + auto result = writer.Append(invalid, [completion, owned = std::move(lifetime)]( + const fluss::WriteCompletion& notification) { + const auto& completed = notification.result; + completion->Record(completed); + }); + EXPECT_FALSE(result.Ok()); + EXPECT_EQ(released.wait_for(std::chrono::seconds(10)), std::future_status::ready); + EXPECT_TRUE(completion->Results().empty()); + + result = + writer.AppendArrowBatch(nullptr, [completion](const fluss::WriteCompletion& notification) { + const auto& completed = notification.result; + completion->Record(completed); + }); + EXPECT_FALSE(result.Ok()); + EXPECT_TRUE(completion->Results().empty()); + // Both failed submissions must not register a callback. + ASSERT_OK(writer.Append(Row(), [completion](const fluss::WriteCompletion& notification) { + const auto& completed = notification.result; + completion->Record(completed); + })); + ASSERT_TRUE(completion->Await()); + ASSERT_OK(writer.Flush()); +} + +TEST_F(WriteCallbackTest, SameBucketCallbacksFireInSubmissionOrder) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto completion = std::make_shared(); + auto order_mutex = std::make_shared(); + auto order = std::make_shared>(); + // A shared id keeps every record on one bucket, so completion order must + // match submission order. The single callback worker must not reorder them. + constexpr int32_t kWrites = 128; + for (int32_t index = 0; index < kWrites; ++index) { + ASSERT_OK(writer.Append(Row(7), [completion, order_mutex, order, + index](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + { + std::lock_guard lock(*order_mutex); + order->push_back(index); + } + completion->Record(result); + })); + } + ASSERT_TRUE(completion->Await(kWrites)); + ASSERT_OK(writer.Flush()); + auto results = completion->Results(); + ASSERT_EQ(results.size(), static_cast(kWrites)); + for (const auto& result : results) { + EXPECT_OK(result); + } + std::lock_guard lock(*order_mutex); + ASSERT_EQ(order->size(), static_cast(kWrites)); + for (int32_t index = 0; index < kWrites; ++index) { + EXPECT_EQ((*order)[index], index); + } +} + +TEST_F(WriteCallbackTest, BatchedCallbacksSurviveExceptionsAndCoexistWithWait) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto completion = std::make_shared(); + constexpr int count = 1024; + for (int i = 0; i < count; ++i) { + // The same bucket key encourages shared internal batches. + ASSERT_OK( + writer.Append(Row(1), [completion, i](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + if (i % 64 == 0) { + throw std::runtime_error("isolated batch callback exception"); + } + })); + } + fluss::WriteResult pending; + ASSERT_OK(writer.Append(Row(1), pending)); + ASSERT_OK(pending.Wait()); + ASSERT_OK(writer.Flush()); + ASSERT_TRUE(completion->Await(count)); + auto results = completion->Results(); + ASSERT_EQ(results.size(), static_cast(count)); + for (const auto& result : results) { + EXPECT_OK(result); + } +} + +TEST_F(WriteCallbackTest, BatchedServerFailureNotifiesEveryAcceptedDelete) { + CreateTable(true, true); + fluss::UpsertWriter writer; + ASSERT_OK(table_.NewUpsert().CreateWriter(writer)); + fluss::WriteResult initial; + ASSERT_OK(writer.Upsert(Row(), initial)); + ASSERT_OK(initial.Wait()); + auto completion = std::make_shared(); + constexpr int count = 257; + for (int i = 0; i < count; ++i) { + ASSERT_OK(writer.Delete(Row(), [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + })); + } + ASSERT_TRUE(completion->Await(count)); + auto results = completion->Results(); + ASSERT_EQ(results.size(), static_cast(count)); + for (const auto& result : results) { + EXPECT_EQ(result.error_code, fluss::ErrorCode::DELETION_DISABLED_EXCEPTION); + } +} + +TEST_F(WriteCallbackTest, EmptyArrowBatchCallbackIsStillAsynchronous) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + arrow::Int32Builder ids; + arrow::StringBuilder values; + auto batch = arrow::RecordBatch::Make( + arrow::schema({arrow::field("id", arrow::int32()), arrow::field("value", arrow::utf8())}), + 0, {ids.Finish().ValueOrDie(), values.Finish().ValueOrDie()}); + auto completion = std::make_shared(); + ASSERT_OK( + writer.AppendArrowBatch(batch, [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + })); + ASSERT_TRUE(completion->Await()); + ASSERT_EQ(completion->Results().size(), 1u); + EXPECT_OK(completion->Results().front()); + EXPECT_NE(completion->Thread(), std::this_thread::get_id()); +} + +TEST_F(WriteCallbackTest, EmptyCallbacksAreRejectedBeforeSubmission) { + CreateTable(); + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer)); + auto result = writer.Append(Row(), nullptr); + EXPECT_FALSE(result.Ok()); + EXPECT_EQ(result.error_message, "Write callback must not be empty"); + result = writer.AppendArrowBatch(nullptr, fluss::WriteCallback{}); + EXPECT_FALSE(result.Ok()); + EXPECT_EQ(result.error_message, "Write callback must not be empty"); + + fluss::UpsertWriter upsert; + result = upsert.Upsert(Row(), nullptr); + EXPECT_FALSE(result.Ok()); + EXPECT_EQ(result.error_message, "Write callback must not be empty"); + result = upsert.Delete(Row(), nullptr); + EXPECT_FALSE(result.Ok()); + EXPECT_EQ(result.error_message, "Write callback must not be empty"); + ASSERT_OK(writer.Flush()); +} + +TEST_F(WriteCallbackTest, UnavailableWritersDoNotInvokeCallbacks) { + auto completion = std::make_shared(); + auto callback = [completion](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + }; + fluss::AppendWriter append; + fluss::UpsertWriter upsert; + EXPECT_FALSE(append.Append(Row(), callback).Ok()); + EXPECT_FALSE(append.AppendArrowBatch(nullptr, callback).Ok()); + EXPECT_FALSE(upsert.Upsert(Row(), callback).Ok()); + EXPECT_FALSE(upsert.Delete(Row(), callback).Ok()); + EXPECT_TRUE(completion->Results().empty()); +} + +TEST(WriteCallbackBridgeTest, ForwardsErrorAndReleasesCaptures) { + auto completion = std::make_shared(); + auto lifetime = std::make_shared(); + auto released = lifetime->released.get_future(); + fluss::ffi::WriteCallback callback( + [completion, owned = std::move(lifetime)](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + completion->Record(result); + }); + callback.Complete(fluss::ErrorCode::DELETION_DISABLED_EXCEPTION, "Deletion is disabled"); + EXPECT_EQ(released.wait_for(std::chrono::seconds(10)), std::future_status::ready); + auto results = completion->Results(); + ASSERT_EQ(results.size(), 1u); + EXPECT_EQ(results.front().error_code, fluss::ErrorCode::DELETION_DISABLED_EXCEPTION); + EXPECT_EQ(results.front().error_message, "Deletion is disabled"); +} + +TEST(WriteCallbackBridgeTest, ContainsCallbackExceptions) { + auto lifetime = std::make_shared(); + auto released = lifetime->released.get_future(); + fluss::ffi::WriteCallback callback( + [owned = std::move(lifetime)](const fluss::WriteCompletion&) { + throw std::runtime_error("callback failure"); + }); + EXPECT_NO_THROW(callback.Complete(0, "")); + EXPECT_EQ(released.wait_for(std::chrono::seconds(10)), std::future_status::ready); + fluss::ffi::WriteCallback unknown([](const fluss::WriteCompletion&) { throw 42; }); + EXPECT_NO_THROW(unknown.Complete(0, "")); +} + +TEST(WriteCallbackBridgeTest, CapacityRejectsOverflowWithoutDroppingReservations) { + constexpr size_t max_pending_operations = 3; + fluss::ffi::WriteCallbackCapacity capacity(max_pending_operations, 0); + for (size_t i = 0; i < max_pending_operations; ++i) { + ASSERT_OK(capacity.Acquire()); + } + EXPECT_FALSE(capacity.Acquire().Ok()); + capacity.Release(); + ASSERT_OK(capacity.Acquire()); + EXPECT_FALSE(capacity.Acquire().Ok()); + for (size_t i = 0; i < max_pending_operations; ++i) { + capacity.Release(); + } + capacity.AwaitAll(); +} + +TEST(WriteCallbackBridgeTest, CapacityLastsThroughCallbackAndCaptureCleanup) { + auto capacity = std::make_shared(1, 0); + // This deleter runs after the user callback returns, but before its slot is returned. + auto capture = std::shared_ptr(new int(0), [capacity](int* value) { + EXPECT_FALSE(capacity->Acquire().Ok()); + delete value; + }); + fluss::ffi::WriteCallback callback( + [capacity, owned = std::move(capture)](const fluss::WriteCompletion&) { + EXPECT_FALSE(capacity->Acquire().Ok()); + throw std::runtime_error("callback failure"); + }); + ASSERT_OK(callback.Reserve(capacity)); + EXPECT_FALSE(capacity->Acquire().Ok()); + callback.Complete(0, ""); + // Complete must release the slot even while its Rust-owned wrapper is still alive. + ASSERT_OK(capacity->Acquire()); + capacity->Release(); +} + +TEST(WriteCallbackBridgeTest, UnsubmittedCallbackReturnsCapacityOnException) { + auto capacity = std::make_shared(1, 0); + int calls = 0; + try { + fluss::ffi::WriteCallback callback([&](const fluss::WriteCompletion&) { ++calls; }); + ASSERT_OK(callback.Reserve(capacity)); + throw std::bad_alloc(); + } catch (const std::bad_alloc&) { + } + EXPECT_EQ(calls, 0); + ASSERT_OK(capacity->Acquire()); + capacity->Release(); +} + +TEST(WriteCallbackBridgeTest, ReservationOutlivesWriterOwnership) { + auto capacity = std::make_shared(1, 0); + std::weak_ptr weak = capacity; + fluss::ffi::WriteCallback callback([](const fluss::WriteCompletion&) {}); + ASSERT_OK(callback.Reserve(capacity)); + capacity.reset(); + EXPECT_FALSE(weak.expired()); + callback.Complete(0, ""); + EXPECT_TRUE(weak.expired()); +} + +TEST(WriteCallbackBridgeTest, CapacityTimeoutDoesNotDiscardAcceptedCallback) { + auto capacity = std::make_shared(1, 25); + int calls = 0; + fluss::ffi::WriteCallback accepted([&](const fluss::WriteCompletion&) { ++calls; }); + ASSERT_OK(accepted.Reserve(capacity)); + auto start = std::chrono::steady_clock::now(); + auto result = capacity->Acquire(); + EXPECT_FALSE(result.Ok()); + EXPECT_NE(result.error_message.find("Timed out"), std::string::npos); + EXPECT_GE(std::chrono::steady_clock::now() - start, std::chrono::milliseconds(25)); + EXPECT_EQ(calls, 0); + accepted.Complete(0, ""); + EXPECT_EQ(calls, 1); + ASSERT_OK(capacity->Acquire()); + capacity->Release(); +} + +TEST(WriteCallbackBridgeTest, WaitingSubmitterResumesAfterCompletion) { + auto capacity = std::make_shared(1, 5000); + fluss::ffi::WriteCallback accepted([](const fluss::WriteCompletion&) {}); + ASSERT_OK(accepted.Reserve(capacity)); + std::promise started; + auto waiter = std::async(std::launch::async, [&] { + started.set_value(); + auto result = capacity->Acquire(); + if (result.Ok()) { + capacity->Release(); + } + return result; + }); + started.get_future().wait(); + EXPECT_EQ(waiter.wait_for(std::chrono::milliseconds(25)), std::future_status::timeout); + accepted.Complete(0, ""); + EXPECT_OK(waiter.get()); +} + +TEST(WriteCallbackBridgeTest, CallbackRejectsFullOtherWriterAndRestoresThreadContext) { + auto capacity = std::make_shared(1, 25); + ASSERT_OK(capacity->Acquire()); // Full writer unrelated to the executing callback. + fluss::ffi::WriteCallback callback([capacity](const fluss::WriteCompletion&) { + auto result = capacity->Acquire(); + EXPECT_EQ(result.error_message, "Write callback capacity is full"); + throw 42; + }); + callback.Complete(0, ""); + auto start = std::chrono::steady_clock::now(); + // CallbackScope must restore the thread context, including on exceptions. + EXPECT_NE(capacity->Acquire().error_message.find("Timed out"), std::string::npos); + EXPECT_GE(std::chrono::steady_clock::now() - start, std::chrono::milliseconds(25)); + capacity->Release(); +} + +TEST(WriteCallbackBridgeTest, ConcurrentCapacityReservationsStayBounded) { + constexpr size_t limit = 3; + auto capacity = std::make_shared(limit, 5000); + std::atomic active{0}; + std::atomic completed{0}; + std::vector threads; + for (int i = 0; i < 8; ++i) { + threads.emplace_back([&] { + for (int j = 0; j < 250; ++j) { + fluss::ffi::WriteCallback callback([&](const fluss::WriteCompletion&) { + --active; + ++completed; + }); + ASSERT_OK(callback.Reserve(capacity)); + EXPECT_LE(++active, limit); + std::this_thread::yield(); + callback.Complete(0, ""); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + EXPECT_EQ(active.load(), 0u); + EXPECT_EQ(completed.load(), 2000u); +} + +TEST(WriteCallbackBridgeTest, FlushRejectsCallbackReentryBeforeTouchingEitherWriter) { + fluss::AppendWriter append; + fluss::UpsertWriter upsert; + EXPECT_FALSE(fluss::ffi::WriteCallbackCapacity::InCallback()); + fluss::ffi::WriteCallback callback([&](const fluss::WriteCompletion&) { + EXPECT_TRUE(fluss::ffi::WriteCallbackCapacity::InCallback()); + EXPECT_EQ(append.Flush().error_message, "Flush cannot be called from a write callback"); + EXPECT_EQ(upsert.Flush().error_message, "Flush cannot be called from a write callback"); + throw std::runtime_error("restore callback context after exception"); + }); + callback.Complete(0, ""); + EXPECT_FALSE(fluss::ffi::WriteCallbackCapacity::InCallback()); + EXPECT_EQ(append.Flush().error_message, "AppendWriter not available"); + EXPECT_EQ(upsert.Flush().error_message, "UpsertWriter not available"); +} + +TEST_F(WriteCallbackTest, RejectsZeroCapacityWithoutCreatingWriter) { + CreateTable(true); + fluss::WriteCallbackOptions options; + EXPECT_EQ(options.max_pending_operations, 262144u); + options.max_pending_operations = 0; + fluss::AppendWriter append; + fluss::UpsertWriter upsert; + auto appended = table_.NewAppend().CreateWriter(append, options); + auto upserted = table_.NewUpsert().CreateWriter(upsert, options); + EXPECT_EQ(appended.error_message, "max_pending_operations must be greater than zero"); + EXPECT_EQ(upserted.error_message, "max_pending_operations must be greater than zero"); + EXPECT_FALSE(append.Available()); + EXPECT_FALSE(upsert.Available()); +} + +TEST_F(WriteCallbackTest, ConfiguredCapacityBlocksUntilCallbackFinishes) { + CreateTable(); + fluss::WriteCallbackOptions options; + options.max_pending_operations = 1; + fluss::AppendWriter writer; + ASSERT_OK(table_.NewAppend().CreateWriter(writer, options)); + auto gate = std::make_shared>(); + auto resume = gate->get_future().share(); + auto started = std::make_shared>(); + auto ready = started->get_future(); + ASSERT_OK(writer.Append(Row(), [started, resume](const fluss::WriteCompletion&) { + started->set_value(); + resume.wait_for(std::chrono::seconds(10)); + })); + ASSERT_EQ(ready.wait_for(std::chrono::seconds(10)), std::future_status::ready); + std::promise submitting; + auto submitted = std::async(std::launch::async, [&] { + submitting.set_value(); + return writer.Append(Row(2), [](const fluss::WriteCompletion&) {}); + }); + submitting.get_future().wait(); + EXPECT_EQ(submitted.wait_for(std::chrono::milliseconds(25)), std::future_status::timeout); + gate->set_value(); + ASSERT_OK(submitted.get()); + ASSERT_OK(writer.Flush()); +} diff --git a/fluss-rust/bindings/cpp/test/test_write_callback_allocation.cpp b/fluss-rust/bindings/cpp/test/test_write_callback_allocation.cpp new file mode 100644 index 00000000000..19cbe5c90f3 --- /dev/null +++ b/fluss-rust/bindings/cpp/test/test_write_callback_allocation.cpp @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include + +#include +#include + +#include "write_callback.hpp" + +namespace { +// Fault injection is confined to this test thread; production has no allocation hook. +thread_local bool fail_next_allocation = false; +} // namespace + +void* operator new(std::size_t size) { + if (std::exchange(fail_next_allocation, false)) { + throw std::bad_alloc(); + } + if (void* value = std::malloc(size ? size : 1)) { + return value; + } + throw std::bad_alloc(); +} + +void operator delete(void* value) noexcept { std::free(value); } +void operator delete(void* value, std::size_t) noexcept { std::free(value); } + +TEST(WriteCallbackBridgeTest, ErrorTextAllocationFailureStillCompletesAndReleasesCapture) { + const std::string message(4096, 'x'); // Exceeds small-string capacity. + auto lifetime = std::make_shared(42); + std::weak_ptr weak = lifetime; + fluss::Result observed; + int calls = 0; + fluss::ffi::WriteCallback callback( + [&, owned = std::move(lifetime)](const fluss::WriteCompletion& notification) { + const auto& result = notification.result; + ++calls; + observed = result; + }); + auto capacity = std::make_shared(1, 0); + ASSERT_TRUE(callback.Reserve(capacity).Ok()); + const rust::Str text(message); + fail_next_allocation = true; + callback.Complete(fluss::ErrorCode::DELETION_DISABLED_EXCEPTION, text); + const bool allocation_was_attempted = !std::exchange(fail_next_allocation, false); + + EXPECT_TRUE(allocation_was_attempted); + EXPECT_EQ(calls, 1); + EXPECT_EQ(observed.error_code, fluss::ErrorCode::DELETION_DISABLED_EXCEPTION); + EXPECT_TRUE(observed.error_message.empty()); + EXPECT_TRUE(weak.expired()); + ASSERT_TRUE(capacity->Acquire().Ok()); + capacity->Release(); +} diff --git a/fluss-rust/crates/fluss/src/client/table/append.rs b/fluss-rust/crates/fluss/src/client/table/append.rs index 790321100fd..62eb8e32182 100644 --- a/fluss-rust/crates/fluss/src/client/table/append.rs +++ b/fluss-rust/crates/fluss/src/client/table/append.rs @@ -29,6 +29,7 @@ use bytes::Bytes; use parking_lot::Mutex; use std::collections::HashMap; use std::sync::Arc; +use std::time::Instant; pub struct TableAppend { table_path: Arc, @@ -137,6 +138,18 @@ impl AppendWriter { /// A [`WriteResultFuture`] that can be awaited to wait for server acknowledgment, /// or dropped for fire-and-forget behavior (use `flush()` to ensure delivery). pub fn append(&self, row: &R) -> Result { + self.append_with_deadline(row, None) + } + + /// Internal: bounds the buffer-memory wait by `deadline` so a language binding can share + /// one `client.writer.buffer.wait-timeout` budget with callback admission waits. Not a + /// public deadline API; use [`Self::append`], which passes `None` and defers to the config. + #[doc(hidden)] + pub fn append_with_deadline( + &self, + row: &R, + deadline: Option, + ) -> Result { self.check_field_count(row)?; let physical_table_path = Arc::new(get_physical_path( &self.table_path, @@ -153,7 +166,8 @@ impl AppendWriter { self.table_info.schema_id, row, ) - .with_bucket_key(bucket_key); + .with_bucket_key(bucket_key) + .with_submit_deadline(deadline); let result_handle = self.writer_client.send(&record)?; Ok(WriteResultFuture::new(result_handle)) } @@ -171,6 +185,18 @@ impl AppendWriter { /// A [`WriteResultFuture`] that can be awaited to wait for server acknowledgment, /// or dropped for fire-and-forget behavior (use `flush()` to ensure delivery). pub fn append_arrow_batch(&self, batch: RecordBatch) -> Result { + self.append_arrow_batch_with_deadline(batch, None) + } + + /// Internal: bounds the buffer-memory wait by `deadline` so a language binding can share + /// one `client.writer.buffer.wait-timeout` budget with callback admission waits. Not a + /// public deadline API; use [`Self::append_arrow_batch`], which passes `None`. + #[doc(hidden)] + pub fn append_arrow_batch_with_deadline( + &self, + batch: RecordBatch, + deadline: Option, + ) -> Result { if batch.num_rows() == 0 { // Nothing to write; also avoids a keyless send to a bucket-key table. return Ok(WriteResultFuture::join(Vec::new())); @@ -193,7 +219,7 @@ impl AppendWriter { }; let Some(router) = self.bucket_router.as_ref() else { - return self.send_arrow_batch(batch, physical_table_path, None); + return self.send_arrow_batch(batch, physical_table_path, None, deadline); }; // Group rows by bucket, keeping one key per bucket (it hashes back there). @@ -215,7 +241,7 @@ impl AppendWriter { if groups.len() == 1 { let (_, (_, rep_key)) = groups.into_iter().next().unwrap(); - return self.send_arrow_batch(batch, physical_table_path, Some(rep_key)); + return self.send_arrow_batch(batch, physical_table_path, Some(rep_key), deadline); } let mut handles = Vec::with_capacity(groups.len()); @@ -227,7 +253,8 @@ impl AppendWriter { self.table_info.schema_id, sub_batch, ) - .with_bucket_key(Some(rep_key)); + .with_bucket_key(Some(rep_key)) + .with_submit_deadline(deadline); handles.push(self.writer_client.send(&record)?); } Ok(WriteResultFuture::join(handles)) @@ -238,6 +265,7 @@ impl AppendWriter { batch: RecordBatch, physical_table_path: Arc, bucket_key: Option, + deadline: Option, ) -> Result { let record = WriteRecord::for_append_record_batch( Arc::clone(&self.table_info), @@ -245,7 +273,8 @@ impl AppendWriter { self.table_info.schema_id, batch, ) - .with_bucket_key(bucket_key); + .with_bucket_key(bucket_key) + .with_submit_deadline(deadline); let result_handle = self.writer_client.send(&record)?; Ok(WriteResultFuture::new(result_handle)) } diff --git a/fluss-rust/crates/fluss/src/client/table/upsert.rs b/fluss-rust/crates/fluss/src/client/table/upsert.rs index 28dce4ee799..efb8bf48151 100644 --- a/fluss-rust/crates/fluss/src/client/table/upsert.rs +++ b/fluss-rust/crates/fluss/src/client/table/upsert.rs @@ -23,6 +23,7 @@ use crate::row::InternalRow; use crate::row::encode::{KeyEncoder, KeyEncoderFactory, RowEncoder, RowEncoderFactory}; use crate::row::field_getter::FieldGetter; use std::sync::Arc; +use std::time::Instant; use crate::client::table::partition_getter::{PartitionGetter, get_physical_path}; use bitvec::prelude::bitvec; @@ -347,6 +348,18 @@ impl UpsertWriter { /// A [`WriteResultFuture`] that can be awaited to wait for server acknowledgment, /// or dropped for fire-and-forget behavior (use `flush()` to ensure delivery). pub fn upsert(&self, row: &R) -> Result { + self.upsert_with_deadline(row, None) + } + + /// Internal: bounds the buffer-memory wait by `deadline` so a language binding can share + /// one `client.writer.buffer.wait-timeout` budget with callback admission waits. Not a + /// public deadline API; use [`Self::upsert`], which passes `None` and defers to the config. + #[doc(hidden)] + pub fn upsert_with_deadline( + &self, + row: &R, + deadline: Option, + ) -> Result { self.check_field_count(row)?; let (key, bucket_key) = self.get_keys(row)?; @@ -369,7 +382,8 @@ impl UpsertWriter { self.write_format, self.target_columns.clone(), Some(row_bytes), - ); + ) + .with_submit_deadline(deadline); let result_handle = self.writer_client.send(&write_record)?; Ok(WriteResultFuture::new(result_handle)) @@ -388,6 +402,18 @@ impl UpsertWriter { /// A [`WriteResultFuture`] that can be awaited to wait for server acknowledgment, /// or dropped for fire-and-forget behavior (use `flush()` to ensure delivery). pub fn delete(&self, row: &R) -> Result { + self.delete_with_deadline(row, None) + } + + /// Internal: bounds the buffer-memory wait by `deadline` so a language binding can share + /// one `client.writer.buffer.wait-timeout` budget with callback admission waits. Not a + /// public deadline API; use [`Self::delete`], which passes `None` and defers to the config. + #[doc(hidden)] + pub fn delete_with_deadline( + &self, + row: &R, + deadline: Option, + ) -> Result { self.check_field_count(row)?; let (key, bucket_key) = self.get_keys(row)?; @@ -405,7 +431,8 @@ impl UpsertWriter { self.write_format, self.target_columns.clone(), None, - ); + ) + .with_submit_deadline(deadline); let result_handle = self.writer_client.send(&write_record)?; Ok(WriteResultFuture::new(result_handle)) diff --git a/fluss-rust/crates/fluss/src/client/write/accumulator.rs b/fluss-rust/crates/fluss/src/client/write/accumulator.rs index dc6b2c9341c..72c22fb5f61 100644 --- a/fluss-rust/crates/fluss/src/client/write/accumulator.rs +++ b/fluss-rust/crates/fluss/src/client/write/accumulator.rs @@ -70,7 +70,20 @@ impl MemoryLimiter { /// Try to acquire `size` bytes. Blocks until memory is available, /// the timeout expires, or the limiter is closed. /// Returns a `MemoryPermit` on success. + #[cfg(test)] pub fn acquire(self: &Arc, size: usize) -> Result { + self.acquire_within(size, None) + } + + /// Like [`acquire`], but bounds the wait by `deadline` when provided instead of + /// the limiter's configured `wait_timeout`. A deadline already in the past makes + /// this non-blocking (fail fast if memory is unavailable), which callers use to + /// share the buffer wait budget with callback admission, not bound the whole API call. + pub fn acquire_within( + self: &Arc, + size: usize, + deadline: Option, + ) -> Result { if self.closed.load(Ordering::Acquire) { return Err(Error::WriterClosed { message: "Memory limiter is closed".to_string(), @@ -87,7 +100,7 @@ impl MemoryLimiter { } let mut used = self.state.lock(); - let deadline = Instant::now() + self.wait_timeout; + let deadline = deadline.unwrap_or_else(|| Instant::now() + self.wait_timeout); while *used + size > self.max_memory { self.waiting_count.fetch_add(1, Ordering::Relaxed); let result = self.cond.wait_until(&mut used, deadline); @@ -101,10 +114,9 @@ impl MemoryLimiter { if result.timed_out() && *used + size > self.max_memory { return Err(Error::BufferExhausted { message: format!( - "Failed to allocate {} bytes for write batch within {}ms. \ + "Failed to allocate {} bytes for write batch within the buffer wait budget. \ {} of {} bytes in use, {} threads waiting.", size, - self.wait_timeout.as_millis(), *used, self.max_memory, self.waiting_count.load(Ordering::Relaxed), @@ -412,7 +424,9 @@ impl RecordAccumulator { let batch_size = dynamic_target.unwrap_or(self.config.writer_batch_size as usize); let record_size = record.estimated_record_size(); let alloc_size = batch_size.max(record_size); - let permit = self.memory_limiter.acquire(alloc_size)?; + let permit = self + .memory_limiter + .acquire_within(alloc_size, record.submit_deadline)?; // Re-acquire dq lock after memory is available let mut dq_guard = dq.lock(); @@ -2346,6 +2360,51 @@ mod tests { assert!(elapsed >= Duration::from_millis(80)); // allow some timing slack } + #[test] + fn test_memory_limiter_acquire_within_bounds_wait_by_deadline() { + // Writer default wait is effectively unbounded; a caller deadline must win. + let limiter = Arc::new(MemoryLimiter::new(1024, Duration::from_secs(3600))); + let _permit = limiter.acquire(1024).unwrap(); + + let start = Instant::now(); + let result = limiter.acquire_within(512, Some(Instant::now() + Duration::from_millis(100))); + let elapsed = start.elapsed(); + + // Returns within the caller budget, not the 1h configured wait_timeout. + assert!(matches!(result.unwrap_err(), Error::BufferExhausted { .. })); + assert!(elapsed >= Duration::from_millis(80)); + assert!(elapsed < Duration::from_secs(2)); + } + + #[test] + fn test_memory_limiter_acquire_within_past_deadline_is_nonblocking() { + // A deadline already in the past = try semantics (zero submit budget). + let limiter = Arc::new(MemoryLimiter::new(1024, Duration::from_secs(3600))); + let _permit = limiter.acquire(1024).unwrap(); + + let start = Instant::now(); + let result = limiter.acquire_within(512, Some(Instant::now() - Duration::from_millis(1))); + let elapsed = start.elapsed(); + + assert!(matches!(result.unwrap_err(), Error::BufferExhausted { .. })); + assert!(elapsed < Duration::from_millis(50)); + } + + #[test] + fn test_memory_limiter_acquire_within_succeeds_when_capacity_available() { + // A bounded deadline must not prevent an allocation that fits right away. + let limiter = Arc::new(MemoryLimiter::new(1024, Duration::from_secs(3600))); + + let start = Instant::now(); + let permit = limiter + .acquire_within(512, Some(Instant::now() + Duration::from_millis(100))) + .unwrap(); + assert!(start.elapsed() < Duration::from_millis(50)); + assert_eq!(*limiter.state.lock(), 512); + drop(permit); + assert_eq!(*limiter.state.lock(), 0); + } + #[test] fn test_memory_limiter_close_fails_immediately() { let limiter = Arc::new(MemoryLimiter::new(1024, Duration::from_secs(60))); diff --git a/fluss-rust/crates/fluss/src/client/write/broadcast.rs b/fluss-rust/crates/fluss/src/client/write/broadcast.rs index 9e00403586f..765d76ae068 100644 --- a/fluss-rust/crates/fluss/src/client/write/broadcast.rs +++ b/fluss-rust/crates/fluss/src/client/write/broadcast.rs @@ -15,7 +15,8 @@ // specific language governing permissions and limitations // under the License. -use parking_lot::RwLock; +use parking_lot::{Mutex, RwLock}; +use std::panic::{AssertUnwindSafe, catch_unwind}; use std::sync::Arc; use thiserror::Error; use tokio::sync::Notify; @@ -24,6 +25,43 @@ pub type Result = std::result::Result; pub type BatchWriteResult = Result<(), Error>; +type Callback = Box) + Send + 'static>; +type Dispatcher = fn(CompletionBatch); + +/// An owned set of callbacks sharing one published result. +/// +/// Binding executors enqueue this batch and run it off the I/O thread. +/// User callbacks are never invoked by the broadcast itself. +#[doc(hidden)] +pub struct CompletionBatch { + result: Arc>, + callbacks: Vec>, +} + +impl CompletionBatch { + /// Execute every callback, isolating panics so later callbacks still run. + pub fn run(self) { + for callback in self.callbacks { + if catch_unwind(AssertUnwindSafe(|| callback(&self.result))).is_err() { + log::error!("Write completion callback panicked"); + } + } + } +} + +struct CallbackGroup { + dispatch: Dispatcher, + callbacks: Vec>, +} + +impl std::fmt::Debug for CallbackGroup { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CallbackGroup") + .field("len", &self.callbacks.len()) + .finish() + } +} + #[derive(Debug, Error, Clone, PartialEq, Eq)] pub enum Error { #[error("BroadcastOnce dropped")] @@ -42,7 +80,33 @@ pub struct BroadcastOnceReceiver { impl BroadcastOnceReceiver { /// Returns `Some(_)` if data has been produced pub fn peek(&self) -> Option> { - self.shared.data.read().clone() + self.shared.data.read().as_deref().cloned() + } + + /// Register under the result read lock. Late registrations join the same + /// dispatch queue, so they cannot overtake callbacks awaiting dispatch. + pub(crate) fn subscribe(&self, callback: Callback, dispatch: Dispatcher) { + let data = self.shared.data.read(); + let result = data.as_ref().map(Arc::clone); + { + let mut state = self.shared.callbacks.lock(); + if let Some(group) = state + .groups + .last_mut() + .filter(|group| std::ptr::fn_addr_eq(group.dispatch, dispatch)) + { + group.callbacks.push(callback); + } else { + state.groups.push(CallbackGroup { + dispatch, + callbacks: vec![callback], + }); + } + } + drop(data); + if let Some(result) = result { + self.shared.dispatch_callbacks(result, false); + } } /// Waits for [`BroadcastOnce::broadcast`] to be called or returns an error @@ -63,18 +127,78 @@ impl BroadcastOnceReceiver { /// Used by `abort_batches` to fail in-flight handles that can't be /// reached through `WriteBatch::complete`. pub(crate) fn fail(&self, error: Error) { - let mut data = self.shared.data.write(); - if data.is_none() { - *data = Some(Err(error)); - self.shared.notify.notify_waiters(); + let result = Arc::new(Err(error)); + { + let mut data = self.shared.data.write(); + if data.is_some() { + return; + } + *data = Some(Arc::clone(&result)); } + self.shared.notify_completion(result); } } #[derive(Debug)] struct Shared { - data: RwLock>>, + data: RwLock>>>, notify: Notify, + callbacks: Mutex>, +} + +#[derive(Debug)] +struct CallbackState { + groups: Vec>, + dispatching: bool, + publication_notified: bool, +} + +impl Default for CallbackState { + fn default() -> Self { + Self { + groups: Vec::new(), + dispatching: false, + publication_notified: false, + } + } +} + +impl Shared { + fn notify_completion(&self, result: Arc>) { + self.notify.notify_waiters(); + self.dispatch_callbacks(result, true); + } + + fn dispatch_callbacks(&self, result: Arc>, publishing: bool) { + let mut state = self.callbacks.lock(); + state.publication_notified |= publishing; + // The publishing thread must own the first drain. A late subscriber + // must not steal it while publication is between storing the result + // and notifying: the publisher could otherwise return and complete + // the next batch before this batch has actually been dispatched. + if !state.publication_notified || state.dispatching { + return; + } + state.dispatching = true; + loop { + let groups = std::mem::take(&mut state.groups); + if groups.is_empty() { + state.dispatching = false; + return; + } + drop(state); + // Exactly one drainer invokes dispatchers, outside every lock. + // Registrations during dispatch (including reentrant ones) queue + // behind this batch rather than dispatching ahead of it. + for group in groups { + (group.dispatch)(CompletionBatch { + result: Arc::clone(&result), + callbacks: group.callbacks, + }); + } + state = self.callbacks.lock(); + } + } } #[derive(Debug)] @@ -94,6 +218,7 @@ where shared: Arc::new(Shared { data: Default::default(), notify: Default::default(), + callbacks: Default::default(), }), } } @@ -110,11 +235,16 @@ impl BroadcastOnce { /// Broadcast a value to all [`BroadcastOnceReceiver`] handles pub fn broadcast(&self, r: T) { - let mut locked = self.shared.data.write(); - assert!(locked.is_none(), "double publish"); - - *locked = Some(Ok(r)); - self.shared.notify.notify_waiters(); + let result = Arc::new(Ok(r)); + { + let mut locked = self.shared.data.write(); + assert!(locked.is_none(), "double publish"); + *locked = Some(Arc::clone(&result)); + } + // Woken receivers immediately read the result. Publish it and release + // the write lock before waking them, rather than make them contend + // with the notification loop for the same lock. + self.shared.notify_completion(result); } } @@ -123,11 +253,336 @@ where T: Send + Sync, { fn drop(&mut self) { - let mut data = self.shared.data.write(); - if data.is_none() { - log::warn!("BroadcastOnce dropped without producing"); - *data = Some(Err(Error::Dropped)); - self.shared.notify.notify_waiters(); + let result = { + let mut data = self.shared.data.write(); + if data.is_some() { + return; + } + let result = Arc::new(Err(Error::Dropped)); + *data = Some(Arc::clone(&result)); + result + }; + log::warn!("BroadcastOnce dropped without producing"); + self.shared.notify_completion(result); + } +} + +#[cfg(test)] +mod tests { + use super::{BroadcastOnce, CompletionBatch, Error, Shared}; + use std::future::Future; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::task::{Context, Poll, Wake, Waker}; + use std::time::Duration; + + #[test] + fn test_callbacks_share_one_dispatch_and_late_registration_is_delivered() { + use std::sync::atomic::AtomicUsize; + static DISPATCHES: AtomicUsize = AtomicUsize::new(0); + fn dispatch(batch: CompletionBatch) { + DISPATCHES.fetch_add(1, Ordering::SeqCst); + batch.run(); } + let broadcast = BroadcastOnce::default(); + let receiver = broadcast.receiver(); + let (tx, rx) = std::sync::mpsc::channel(); + for index in 0..1000 { + let tx = tx.clone(); + receiver.subscribe( + Box::new(move |result| tx.send((index, result.clone())).unwrap()), + dispatch, + ); + } + assert_eq!(DISPATCHES.load(Ordering::SeqCst), 0); + broadcast.broadcast(42); + assert_eq!(DISPATCHES.load(Ordering::SeqCst), 1); + let late_tx = tx.clone(); + receiver.subscribe( + Box::new(move |result| late_tx.send((1000, result.clone())).unwrap()), + dispatch, + ); + assert_eq!(DISPATCHES.load(Ordering::SeqCst), 2); + drop(tx); + let mut results: Vec<_> = rx.into_iter().collect(); + results.sort_by_key(|(index, _)| *index); + assert_eq!(results, (0..1001).map(|i| (i, Ok(42))).collect::>()); + } + + #[test] + fn test_panics_do_not_drop_remaining_callbacks() { + fn dispatch(batch: CompletionBatch) { + batch.run(); + } + let broadcast = BroadcastOnce::default(); + let (tx, rx) = std::sync::mpsc::channel(); + broadcast + .receiver() + .subscribe(Box::new(|_| panic!("isolated callback")), dispatch); + for i in 0..130 { + let tx = tx.clone(); + broadcast + .receiver() + .subscribe(Box::new(move |_| tx.send(i).unwrap()), dispatch); + } + broadcast.broadcast(42); + drop(tx); + assert_eq!( + rx.into_iter().collect::>(), + (0..130).collect::>() + ); + } + + #[test] + fn test_callback_registration_races_all_terminal_paths() { + for mode in 0..3 { + for _ in 0..16 { + let broadcast = BroadcastOnce::::default(); + let receiver = broadcast.receiver(); + let (tx, rx) = std::sync::mpsc::channel(); + let barrier = Arc::new(std::sync::Barrier::new(5)); + let mut threads = Vec::new(); + for producer in 0..4 { + let receiver = receiver.clone(); + let tx = tx.clone(); + let barrier = Arc::clone(&barrier); + threads.push(std::thread::spawn(move || { + barrier.wait(); + for i in 0..32 { + let tx = tx.clone(); + receiver.subscribe( + Box::new(move |r| tx.send((producer * 32 + i, r.clone())).unwrap()), + CompletionBatch::run, + ); + } + })); + } + barrier.wait(); + let expected = match mode { + 0 => { + broadcast.broadcast(42); + Ok(42) + } + 1 => { + receiver.fail(Error::Client { + message: "abort".into(), + }); + Err(Error::Client { + message: "abort".into(), + }) + } + _ => { + drop(broadcast); + Err(Error::Dropped) + } + }; + for thread in threads { + thread.join().unwrap(); + } + drop(tx); + let mut results: Vec<_> = rx.into_iter().collect(); + results.sort_by_key(|(index, _)| *index); + assert_eq!( + results, + (0..128).map(|i| (i, expected.clone())).collect::>() + ); + assert_eq!(receiver.peek(), Some(expected)); + } + } + } + + #[test] + fn test_dispatch_runs_outside_locks_and_failure_cannot_complete_twice() { + let broadcast = BroadcastOnce::::default(); + let receiver = broadcast.receiver(); + let nested = receiver.clone(); + let (tx, rx) = std::sync::mpsc::channel(); + receiver.subscribe( + Box::new(move |result| { + assert_eq!(nested.peek(), Some(result.clone())); + // A synchronous test dispatcher deliberately reenters registration. + nested.subscribe( + Box::new(move |r| tx.send(r.clone()).unwrap()), + CompletionBatch::run, + ); + }), + CompletionBatch::run, + ); + broadcast.broadcast(42); + receiver.fail(Error::Dropped); + drop(broadcast); + assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), Ok(42)); + assert!(matches!( + rx.try_recv(), + Err(std::sync::mpsc::TryRecvError::Disconnected) + )); + } + + #[test] + fn test_late_registration_cannot_overtake_published_callbacks() { + let broadcast = BroadcastOnce::::default(); + let receiver = broadcast.receiver(); + let (tx, rx) = std::sync::mpsc::channel(); + let first = tx.clone(); + receiver.subscribe( + Box::new(move |_| first.send(1).unwrap()), + CompletionBatch::run, + ); + // Pause publication exactly between making data visible and notifying. + let result = Arc::new(Ok(42)); + *receiver.shared.data.write() = Some(Arc::clone(&result)); + receiver.subscribe(Box::new(move |_| tx.send(2).unwrap()), CompletionBatch::run); + // Merely observing the published data must not take ownership of the + // publisher's first dispatch, including its earlier registered callback. + assert!(matches!( + rx.try_recv(), + Err(std::sync::mpsc::TryRecvError::Empty) + )); + receiver.shared.notify_completion(result); + assert_eq!(rx.into_iter().collect::>(), vec![1, 2]); + } + + #[test] + fn test_reentrant_registration_runs_after_existing_callbacks() { + let broadcast = BroadcastOnce::::default(); + let receiver = broadcast.receiver(); + let nested = receiver.clone(); + let (tx, rx) = std::sync::mpsc::channel(); + let first = tx.clone(); + receiver.subscribe( + Box::new(move |_| { + first.send(1).unwrap(); + nested.subscribe( + Box::new(move |_| first.send(3).unwrap()), + CompletionBatch::run, + ); + }), + CompletionBatch::run, + ); + receiver.subscribe(Box::new(move |_| tx.send(2).unwrap()), CompletionBatch::run); + broadcast.broadcast(42); + assert_eq!(rx.into_iter().collect::>(), vec![1, 2, 3]); + } + + struct InspectOnWake { + shared: Arc>, + woke: AtomicBool, + result_readable: AtomicBool, + } + + impl InspectOnWake { + fn inspect(&self) { + self.woke.store(true, Ordering::SeqCst); + let readable = self + .shared + .data + .try_read() + .is_some_and(|data| data.is_some()); + self.result_readable.store(readable, Ordering::SeqCst); + } + } + + impl Wake for InspectOnWake { + fn wake(self: Arc) { + self.inspect(); + } + + fn wake_by_ref(self: &Arc) { + self.inspect(); + } + } + + fn assert_unlocked_on_notification( + action: impl FnOnce(BroadcastOnce), + expected: super::Result, + ) { + let broadcast = BroadcastOnce::::default(); + let receiver = broadcast.receiver(); + let probe = Arc::new(InspectOnWake { + shared: Arc::clone(&receiver.shared), + woke: AtomicBool::new(false), + result_readable: AtomicBool::new(false), + }); + let waker = Waker::from(Arc::clone(&probe)); + let mut context = Context::from_waker(&waker); + let mut future = Box::pin(receiver.receive()); + assert!(future.as_mut().poll(&mut context).is_pending()); + action(broadcast); + assert!(probe.woke.load(Ordering::SeqCst)); + assert!(probe.result_readable.load(Ordering::SeqCst)); + assert_eq!(future.as_mut().poll(&mut context), Poll::Ready(expected)); + } + + #[test] + fn test_broadcast_releases_result_lock_before_waking() { + assert_unlocked_on_notification(|broadcast| broadcast.broadcast(42), Ok(42)); + } + + #[test] + fn test_failure_releases_result_lock_before_waking() { + let error = Error::Client { + message: "writer closed".to_string(), + }; + assert_unlocked_on_notification( + |broadcast| broadcast.receiver().fail(error.clone()), + Err(error.clone()), + ); + } + + #[test] + fn test_drop_releases_result_lock_before_waking() { + assert_unlocked_on_notification(drop, Err(Error::Dropped)); + } + + #[test] + fn test_failure_does_not_overwrite_published_result() { + let broadcast = BroadcastOnce::::default(); + let receiver = broadcast.receiver(); + broadcast.broadcast(42); + receiver.fail(Error::Dropped); + assert_eq!(receiver.peek(), Some(Ok(42))); + drop(broadcast); + assert_eq!(receiver.peek(), Some(Ok(42))); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_receivers_observe_results_before_and_after_notification() { + tokio::time::timeout(Duration::from_secs(10), async { + for round in 0..64 { + let broadcast = BroadcastOnce::::default(); + let late = broadcast.receiver(); + let barrier = Arc::new(tokio::sync::Barrier::new(33)); + let mut readers = Vec::new(); + for _ in 0..32 { + let receiver = broadcast.receiver(); + let barrier = Arc::clone(&barrier); + readers.push(tokio::spawn(async move { + barrier.wait().await; + receiver.receive().await + })); + } + barrier.wait().await; + let expected = match round % 3 { + 0 => { + broadcast.broadcast(round); + Ok(round) + } + 1 => { + broadcast.receiver().fail(Error::Dropped); + Err(Error::Dropped) + } + _ => { + drop(broadcast); + Err(Error::Dropped) + } + }; + for reader in readers { + assert_eq!(reader.await.unwrap(), expected); + } + assert_eq!(late.receive().await, expected); + } + }) + .await + .expect("all receivers must finish without a missed notification"); } } diff --git a/fluss-rust/crates/fluss/src/client/write/mod.rs b/fluss-rust/crates/fluss/src/client/write/mod.rs index 37bdbffe3fa..51fd52c224f 100644 --- a/fluss-rust/crates/fluss/src/client/write/mod.rs +++ b/fluss-rust/crates/fluss/src/client/write/mod.rs @@ -32,6 +32,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; +use std::time::Instant; pub(crate) mod broadcast; mod bucket_assigner; @@ -52,6 +53,9 @@ pub struct WriteRecord<'a> { schema_id: i32, write_format: WriteFormat, table_info: Arc, + /// Optional deadline bounding the buffer-memory wait during append. `None` + /// falls back to the writer's configured buffer wait timeout. + submit_deadline: Option, } impl<'a> WriteRecord<'a> { @@ -143,6 +147,7 @@ impl<'a> WriteRecord<'a> { bucket_key: None, schema_id, write_format: WriteFormat::ArrowLog, + submit_deadline: None, } } @@ -159,6 +164,7 @@ impl<'a> WriteRecord<'a> { bucket_key: None, schema_id, write_format: WriteFormat::ArrowLog, + submit_deadline: None, } } @@ -168,6 +174,14 @@ impl<'a> WriteRecord<'a> { self } + /// Sets a submit deadline that bounds how long the buffer-memory wait may block + /// before this record's append fails fast. `None` uses the writer's configured + /// buffer wait timeout. + pub fn with_submit_deadline(mut self, deadline: Option) -> Self { + self.submit_deadline = deadline; + self + } + #[allow(clippy::too_many_arguments)] pub fn for_upsert( table_info: Arc, @@ -186,6 +200,7 @@ impl<'a> WriteRecord<'a> { bucket_key, schema_id, write_format, + submit_deadline: None, } } } @@ -216,6 +231,10 @@ impl ResultHandle { } pub fn result(&self, batch_result: BatchWriteResult) -> Result<(), Error> { + Self::resolve(batch_result) + } + + fn resolve(batch_result: BatchWriteResult) -> Result<(), Error> { batch_result.map_err(|e| match e { client_broadcast::Error::WriteFailed { code, message } => Error::FlussAPIError { api_error: crate::rpc::ApiError { code, message }, @@ -241,9 +260,19 @@ impl ResultHandle { /// This pattern is similar to rdkafka's `DeliveryFuture` and allows for efficient batching /// when users don't need immediate per-record acknowledgment. pub struct WriteResultFuture { - inner: Pin> + Send>>, + state: WriteResultState, +} + +enum WriteResultState { + Single(ResultHandle), + Joined(Vec), + Waiting(Pin> + Send>>), } +/// An opaque group of write completions for language-binding executors. +#[doc(hidden)] +pub type WriteCallbackBatch = broadcast::CompletionBatch; + impl std::fmt::Debug for WriteResultFuture { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("WriteResultFuture").finish_non_exhaustive() @@ -254,30 +283,334 @@ impl WriteResultFuture { /// Create a new WriteResultFuture from a ResultHandle. pub fn new(result_handle: ResultHandle) -> Self { Self { - inner: Box::pin(async move { - let result = result_handle.wait().await?; - result_handle.result(result) - }), + state: WriteResultState::Single(result_handle), } } pub fn join(handles: Vec) -> Self { Self { - inner: Box::pin(async move { - for handle in handles { - let result = handle.wait().await?; - handle.result(result)?; + state: WriteResultState::Joined(handles), + } + } + + /// Register completions directly on their owning batches, without spawning + /// per-record waiting tasks. `dispatch` must reliably enqueue every job, + /// must not panic, and must not run callbacks on the calling/I/O thread. + /// + /// A previously polled future is returned with its callback untouched for + /// an executor to await normally. Dispatch may happen before this returns. + #[doc(hidden)] + pub fn try_on_complete( + self, + callback: C, + dispatch: fn(WriteCallbackBatch), + ) -> std::result::Result<(), (Self, C)> + where + C: FnOnce(Result<(), Error>) + Send + 'static, + { + match self.state { + WriteResultState::Single(handle) => { + handle.receiver.subscribe( + Box::new(move |result| callback(resolve_callback_result(result))), + dispatch, + ); + } + WriteResultState::Joined(handles) if handles.is_empty() => { + // Use the same executor even for an empty Arrow RecordBatch. + let completed = broadcast::BroadcastOnce::default(); + completed.receiver().subscribe( + Box::new(move |result| callback(resolve_callback_result(result))), + dispatch, + ); + completed.broadcast(Ok(())); + } + WriteResultState::Joined(handles) => { + // Preserve join's input-order error semantics, even when batches + // finish out of order. Do not call user code under this mutex. + let count = handles.len(); + let joined = Arc::new(parking_lot::Mutex::new(JoinedCallback { + results: (0..count).map(|_| None).collect(), + next: 0, + callback: Some(callback), + })); + for (index, handle) in handles.into_iter().enumerate() { + let joined = Arc::clone(&joined); + handle.receiver.subscribe( + Box::new(move |result| { + let ready = { + let mut joined = joined.lock(); + joined.complete(index, resolve_callback_result(result)) + }; + if let Some((callback, result)) = ready { + callback(result); + } + }), + dispatch, + ); } - Ok(()) - }), + } + WriteResultState::Waiting(_) => return Err((self, callback)), + } + Ok(()) + } +} + +struct JoinedCallback { + results: Vec>>, + next: usize, + callback: Option, +} + +impl JoinedCallback { + fn complete( + &mut self, + index: usize, + result: Result<(), Error>, + ) -> Option<(C, Result<(), Error>)> { + self.callback.as_ref()?; + self.results[index] = Some(result); + while self.next < self.results.len() { + let result = self.results[self.next].take()?; + self.next += 1; + if result.is_err() || self.next == self.results.len() { + return self.callback.take().map(|callback| (callback, result)); + } } + None } } +fn resolve_callback_result( + result: &client_broadcast::Result, +) -> Result<(), Error> { + let result = result.clone().map_err(|e| Error::UnexpectedError { + message: format!("Fail to wait write result {e:?}"), + source: None, + })?; + ResultHandle::resolve(result) +} + impl Future for WriteResultFuture { type Output = Result<(), Error>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - self.inner.as_mut().poll(cx) + loop { + match &mut self.state { + WriteResultState::Waiting(future) => return future.as_mut().poll(cx), + _ => { + let state = + std::mem::replace(&mut self.state, WriteResultState::Joined(Vec::new())); + self.state = WriteResultState::Waiting(Box::pin(async move { + match state { + WriteResultState::Single(handle) => { + let result = handle.wait().await?; + handle.result(result) + } + WriteResultState::Joined(handles) => { + for handle in handles { + let result = handle.wait().await?; + handle.result(result)?; + } + Ok(()) + } + WriteResultState::Waiting(_) => unreachable!(), + } + })); + } + } + } + } +} + +#[cfg(test)] +mod callback_tests { + use super::*; + use broadcast::BroadcastOnce; + use std::sync::mpsc; + use std::time::Duration; + + fn future(batch: &BroadcastOnce) -> WriteResultFuture { + WriteResultFuture::new(ResultHandle::new(batch.receiver())) + } + + fn register(future: WriteResultFuture) -> mpsc::Receiver> { + let (tx, rx) = mpsc::channel(); + assert!( + future + .try_on_complete(move |r| tx.send(r).unwrap(), WriteCallbackBatch::run) + .is_ok() + ); + rx + } + + #[tokio::test] + async fn test_callback_and_wait_share_result_without_consuming_each_other() { + let batch = BroadcastOnce::default(); + let wait = future(&batch); + let rx = register(future(&batch)); + assert!(matches!(rx.try_recv(), Err(mpsc::TryRecvError::Empty))); + batch.broadcast(Ok(())); + assert!(wait.await.is_ok()); + assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().is_ok()); + assert!(matches!( + rx.try_recv(), + Err(mpsc::TryRecvError::Disconnected) + )); + } + + #[test] + fn test_callbacks_preserve_batch_and_abort_errors() { + let batch = BroadcastOnce::default(); + let rx = register(future(&batch)); + batch.broadcast(Err(client_broadcast::Error::WriteFailed { + code: 57, + message: "Deletion is disabled".into(), + })); + assert!( + matches!(rx.recv().unwrap(), Err(Error::FlussAPIError { api_error }) + if api_error.code == 57 && api_error.message == "Deletion is disabled") + ); + + let batch = BroadcastOnce::default(); + let rx = register(future(&batch)); + batch.receiver().fail(client_broadcast::Error::Client { + message: "abort".into(), + }); + assert!( + rx.recv() + .unwrap() + .unwrap_err() + .to_string() + .contains("abort") + ); + + let batch = BroadcastOnce::default(); + let rx = register(future(&batch)); + drop(batch); + assert!( + rx.recv() + .unwrap() + .unwrap_err() + .to_string() + .contains("Dropped") + ); + } + + #[test] + fn test_join_preserves_input_order_and_short_circuits_error() { + let first = BroadcastOnce::default(); + let second = BroadcastOnce::default(); + let third = BroadcastOnce::default(); + let rx = register(WriteResultFuture::join(vec![ + ResultHandle::new(first.receiver()), + ResultHandle::new(second.receiver()), + ResultHandle::new(third.receiver()), + ])); + second.broadcast(Err(client_broadcast::Error::WriteFailed { + code: 57, + message: "second error".into(), + })); + assert!(matches!(rx.try_recv(), Err(mpsc::TryRecvError::Empty))); + first.broadcast(Ok(())); + assert!( + rx.recv() + .unwrap() + .unwrap_err() + .to_string() + .contains("second error") + ); + // No need to wait for the third batch after the first ordered error. + third.broadcast(Ok(())); + assert!(matches!( + rx.try_recv(), + Err(mpsc::TryRecvError::Disconnected) + )); + } + + #[test] + fn test_join_all_success_empty_and_duplicate_batch_handles() { + assert!( + register(WriteResultFuture::join(Vec::new())) + .recv() + .unwrap() + .is_ok() + ); + let batch = BroadcastOnce::default(); + let rx = register(WriteResultFuture::join(vec![ + ResultHandle::new(batch.receiver()), + ResultHandle::new(batch.receiver()), + ])); + batch.broadcast(Ok(())); + assert!(rx.recv().unwrap().is_ok()); + assert!(matches!( + rx.try_recv(), + Err(mpsc::TryRecvError::Disconnected) + )); + } + + #[test] + fn test_join_first_error_wins_despite_reverse_completion_order() { + let first = BroadcastOnce::default(); + let second = BroadcastOnce::default(); + let rx = register(WriteResultFuture::join(vec![ + ResultHandle::new(first.receiver()), + ResultHandle::new(second.receiver()), + ])); + second.broadcast(Err(client_broadcast::Error::Client { + message: "second".into(), + })); + first.broadcast(Err(client_broadcast::Error::Client { + message: "first".into(), + })); + assert!( + rx.recv() + .unwrap() + .unwrap_err() + .to_string() + .contains("first") + ); + } + + #[test] + fn test_join_concurrent_completions_invoke_callback_once() { + for _ in 0..32 { + let batches: Vec<_> = (0..8).map(|_| BroadcastOnce::default()).collect(); + let rx = register(WriteResultFuture::join( + batches + .iter() + .map(|b| ResultHandle::new(b.receiver())) + .collect(), + )); + std::thread::scope(|scope| { + for batch in &batches { + scope.spawn(move || batch.broadcast(Ok(()))); + } + }); + assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().is_ok()); + assert!(matches!( + rx.try_recv(), + Err(mpsc::TryRecvError::Disconnected) + )); + } + } + + #[tokio::test] + async fn test_polled_future_returns_untouched_callback_for_fallback() { + let batch = BroadcastOnce::default(); + let mut wait = future(&batch); + assert!( + Pin::new(&mut wait) + .poll(&mut Context::from_waker(std::task::Waker::noop())) + .is_pending() + ); + let (tx, rx) = mpsc::channel(); + let registered = + wait.try_on_complete(move |r| tx.send(r).unwrap(), WriteCallbackBatch::run); + let Err((wait, callback)) = registered else { + panic!("polled future must use fallback") + }; + batch.broadcast(Ok(())); + callback(wait.await); + assert!(rx.recv().unwrap().is_ok()); } } diff --git a/fluss-rust/website/docs/user-guide/cpp/api-reference.md b/fluss-rust/website/docs/user-guide/cpp/api-reference.md index ec60464bb63..fa1c03caa33 100644 --- a/fluss-rust/website/docs/user-guide/cpp/api-reference.md +++ b/fluss-rust/website/docs/user-guide/cpp/api-reference.md @@ -25,6 +25,8 @@ Complete API reference for the Fluss C++ client. | `writer_dynamic_batch_size_enabled` | `bool` | `true` | Enable per-table dynamic batch sizing: target grows 10% above 80% fill, shrinks 5% below 50% | | `writer_dynamic_batch_size_min` | `int32_t` | `262144` (256 KB) | Lower bound for the dynamic batch size estimator (ignored when disabled) | | `writer_batch_timeout_ms` | `int64_t` | `100` | Maximum time in ms to wait for a writer batch to fill up before sending | +| `writer_buffer_memory_size` | `size_t` | `67108864` (64 MiB) | Shared write-batch memory budget per Connection, across all tables and writers; not a process RSS limit | +| `writer_buffer_wait_timeout_ms` | `uint64_t` | `UINT64_MAX` | Maximum wait in ms for Rust write-buffer capacity; also provides a shared budget for callback-capacity and buffer waits. `UINT64_MAX` waits indefinitely | | `writer_kv_backpressure_max_throttle_ms` | `uint64_t` | `3000` | Maximum per-bucket KV backpressure throttle in milliseconds | | `writer_bucket_no_key_assigner` | `std::string` | `"sticky"` | Bucket assignment strategy for tables without bucket keys: `"sticky"` or `"round_robin"` | | `scanner_remote_log_prefetch_num` | `size_t` | `4` | Number of remote log segments to prefetch | @@ -130,6 +132,7 @@ Complete API reference for the Fluss C++ client. | Method | Description | |----------------------------------------------|-------------------------| | `CreateWriter(AppendWriter& out) -> Result` | Create an append writer | +| `CreateWriter(AppendWriter& out, const WriteCallbackOptions& options) -> Result` | Create with a per-writer callback operation limit | ## `TableUpsert` @@ -138,6 +141,7 @@ Complete API reference for the Fluss C++ client. | `PartialUpdateByIndex(std::vector column_indices) -> TableUpsert&` | Configure partial update by column indices | | `PartialUpdateByName(std::vector column_names) -> TableUpsert&` | Configure partial update by column names | | `CreateWriter(UpsertWriter& out) -> Result` | Create an upsert writer | +| `CreateWriter(UpsertWriter& out, const WriteCallbackOptions& options) -> Result` | Create with a callback limit shared by upserts and deletes | ## `TableLookup` @@ -188,6 +192,10 @@ are retained. Filters apply to Arrow log scans and are rejected by `CreateBucket |-------------------------------------------------------------|----------------------------------------| | `Append(const GenericRow& row) -> Result` | Append a row (fire-and-forget) | | `Append(const GenericRow& row, WriteResult& out) -> Result` | Append a row with write acknowledgment | +| `Append(const GenericRow& row, WriteCallback callback) -> Result` | Append a row with completion notification | +| `AppendArrowBatch(const std::shared_ptr& batch) -> Result` | Append a batch (fire-and-forget) | +| `AppendArrowBatch(const std::shared_ptr& batch, WriteResult& out) -> Result` | Append a batch with write acknowledgment | +| `AppendArrowBatch(const std::shared_ptr& batch, WriteCallback callback) -> Result` | Append a batch with one completion notification | | `Flush() -> Result` | Flush all pending writes | ## `UpsertWriter` @@ -196,8 +204,10 @@ are retained. Filters apply to Arrow log scans and are rejected by `CreateBucket |-------------------------------------------------------------|-----------------------------------------------| | `Upsert(const GenericRow& row) -> Result` | Upsert a row (fire-and-forget) | | `Upsert(const GenericRow& row, WriteResult& out) -> Result` | Upsert a row with write acknowledgment | +| `Upsert(const GenericRow& row, WriteCallback callback) -> Result` | Upsert a row with completion notification | | `Delete(const GenericRow& row) -> Result` | Delete a row by primary key (fire-and-forget) | | `Delete(const GenericRow& row, WriteResult& out) -> Result` | Delete a row with write acknowledgment | +| `Delete(const GenericRow& row, WriteCallback callback) -> Result` | Delete a row with completion notification | | `Flush() -> Result` | Flush all pending operations | ## `WriteResult` @@ -206,6 +216,318 @@ are retained. Filters apply to Arrow log scans and are rejected by `CreateBucket |--------------------|---------------------------------------------| | `Wait() -> Result` | Wait for server acknowledgment of the write | +## `WriteCallback` + +`WriteCallback` is `std::function`. Pass a function pointer or a +lambda to receive the final write outcome without a `WriteResult` handle or a +call to `Wait()`: + +```cpp +void OnWriteComplete(const fluss::WriteCompletion& completion) { + const auto& completed = completion.result; + if (!completed.Ok()) { + // The record may already have been written. Logging alone is not recovery. + std::cerr << "Write failed: " << completed.error_message << '\n'; + } +} + +auto submitted = writer.Append(row, &OnWriteComplete); +if (!submitted.Ok()) { + // Submission failed; OnWriteComplete will not be called. + std::cerr << "Submission failed: " << submitted.error_message << '\n'; +} +``` + +The immediate return value reports submission status, not acknowledgment. An +empty callback is rejected before submission. During normal operation, each +successfully submitted operation invokes its callback exactly once with its +final success or failure, subject to the lifetime and shutdown requirements below; +`AppendArrowBatch` invokes one callback for the batch, not one per row or bucket. +Submission does not wait for acknowledgment, but may still wait for callback +capacity and then for Rust writer buffer space under backpressure. + +`WriteCompletion` currently contains a `Result result` field. Its reference is valid +only during the callback; copy the result if application workers need it later. + +Callbacks run serially on one process-wide worker in dispatch order, off the I/O +threads. There is no cross-bucket submission-order guarantee. Late registrations +are queued when registered, behind callbacks already awaiting dispatch for that +batch. Keep each callback short and non-blocking. Do not call `Flush()` or `WriteResult::Wait()` inside a +callback, and do not retry synchronously there. Record the outcome and run any +retry in application logic outside the callback. + +### Write guarantees and recovery + +- A successful submission means the operation was accepted, not that it was + written successfully. Check both the immediate return value and the callback + result. A submission error does not register a callback. +- A successful callback reports completion under the configured acknowledgment + policy. It does not strengthen that policy; for example, `writer_acks = "0"` + does not wait for server acknowledgment. +- The SDK handles retryable write errors internally according to its retry + configuration. The callback reports the resulting completion, not each retry + attempt. A failure can be reported when retries are exhausted or an error + cannot be retried; not every failure goes through the configured retry count. +- **A failed callback does not guarantee that the record was not written.** + Even for a single row, the server may have written it before a response was + lost. Other errors can represent a definite rejection. `Result` has no separate + field that distinguishes these outcomes, and `IsRetriable()` is not proof that + nothing was written or that resubmission is duplicate-safe. +- Calling `Append` again is a new operation. SDK idempotence for internal retries + does not deduplicate application resubmissions of the same logical record. + When a callback reports failure, record the outcome and either stop the + pipeline or hand the record to your own retry queue and resubmit from a + separate thread with a bounded policy. Deduplicate resubmissions by application + identifier. A non-retriable error will fail again, so do not blindly resubmit. +- One callback invocation per accepted operation is not an exactly-once delivery + guarantee. Pending notifications are not persisted and can be lost on process + exit or a crash. Applications requiring recovery across restarts must retain + their source records or durable operation state independently of the callback. + A simple approach is to drive writes from a replayable source and advance your + source position or offset only after `Flush()` succeeds and the relevant write + outcomes are confirmed successful, then replay from the + last committed position on restart and deduplicate by identifier. + +`AppendArrowBatch` is not atomic across internal batches or buckets. Its single +callback cannot identify which individual rows succeeded. A failed submission +can also follow partial acceptance without registering a callback; see the +operational limits below before retrying a batch. + +### Callback capacity + +```cpp +fluss::AppendWriter writer; +fluss::WriteCallbackOptions options; +options.max_pending_operations = 262144; // Default, per writer; must be positive. +auto created = table.NewAppend().CreateWriter(writer, options); +// CreateWriter(writer) uses the defaults. Check created before using writer. +// NewUpsert().CreateWriter accepts the same options. +``` + +Each writer independently limits outstanding callback operations with +`max_pending_operations`. This bounds operation count, not bytes, and is not derived +from the schema or `writer_buffer_memory_size`. A callback submission reserves one +slot **before** submitting to Rust +and holds it through user callback execution and capture cleanup. Submission errors and +exceptions return the slot automatically. `Upsert` and `Delete` share their writer's +limit; `AppendArrowBatch` consumes one slot per call, regardless of row count. Moving +a writer transfers its capacity state; already accepted callbacks retain it +independently of the writer's lifetime. + +When callback capacity or the write buffer is full, the submitting thread waits up +to `client.writer.buffer.wait-timeout` for both together. +A capacity rejection returns `CLIENT_ERROR` without submitting any data or +registering a callback; it never discards an accepted notification. Client errors +return false from `IsRetriable()`, including capacity errors. Applications may +reschedule capacity-rejected submissions with backoff, but must not blindly +retry every client error: other submission failures can have different effects, +including partial ArrowBatch acceptance described below. + +Callback capacity admission and buffer backpressure share this wait budget. Zero +makes these waits fail fast if resources are unavailable; it does not make the +entire API call non-blocking. The budget does not bound conversion, scheduling, +network requests, core retries, or callback duration, and does not cancel accepted +writes. Do not hold a mutex needed by callbacks while submitting: a full +writer can wait for those callbacks to finish. + +### Sizing callback capacity and write buffers + +Callback operations and buffered bytes have separate budgets: + +| Setting | Scope | What it limits | +|---------|-------|----------------| +| `max_pending_operations` | Each writer | Operations awaiting write completion, callback execution, or capture cleanup | +| `writer_buffer_memory_size` | Each Connection | Rust write-batch memory accounting shared by all its tables and writers, including writes using `Wait()` or fire-and-forget | + +The Connection buffer accounts for write batches, not callback captures, +application input, or application retry queues. Its accounting is not a precise +bound on actual Arrow builder allocations or process RSS. +Sharing a Connection shares this budget: do not multiply it by the number of +writers, but expect busy or stalled tables to compete for it. Separate Connections +have separate budgets, which must be added when sizing the process or host. + +Callback capacity can bind even while buffer space remains available. The byte +budget is released on batch completion, whereas a callback slot remains held +through callback execution and capture cleanup. With N writers, the sum of their +limits bounds outstanding operations, not a single process-wide limit. +The callback queue does not keep the Rust write-buffer permit after the batch +completes, so a slow callback does not by itself keep acknowledged write bytes +in the Rust buffer. It can still retain callback objects and application captures. +Once a writer reaches `max_pending_operations`, callback-based submissions to that +writer wait or fail according to `client.writer.buffer.wait-timeout`; with the +default unbounded timeout, they can wait indefinitely. + +For an initial estimate, multiply peak accepted operations/second by measured +submit-to-callback-finish latency, including ACK latency and queueing, then allow +headroom for bursts. Check that this count times retained bytes per operation fits +the callback memory budget, including wrapper overhead and captured objects. This +is a sizing estimate, not a worst-case memory guarantee. The default 262144 is a +starting limit, not a value derived from record size. Under sustained overload no +finite queue compensates for callbacks being slower than incoming completions. + +An `AppendArrowBatch` counts as one operation even when it contains many rows, so +large captures need a separate application byte budget. Increase either budget +only after measuring its pressure; increasing one does not increase the other. + +The following explicit settings are a tested high-throughput starting point. They are +**not** new Connection defaults: + +```cpp +fluss::Configuration config; +config.bootstrap_servers = "127.0.0.1:9123"; // Replace with your cluster endpoint. +config.writer_buffer_memory_size = 512ULL * 1024 * 1024; // Per Connection. +config.writer_buffer_wait_timeout_ms = 5000; +config.writer_batch_size = 2 * 1024 * 1024; +config.writer_dynamic_batch_size_min = 1024 * 1024; +config.writer_batch_timeout_ms = 100; +config.writer_request_max_size = 32 * 1024 * 1024; +// Apply config when creating the Connection. +``` + +Keep the 64 MiB Connection default for a small workload unless measurements show +buffer pressure. For sustained high-throughput writes with several writers, +512 MiB per Connection is a tested starting point if the host has sufficient +headroom. Eight such Connections have a 4 GiB write-buffer budget in total, not a +4 GiB RSS limit. More active buckets and tables can retain more concurrent +batches; tune using buffer pressure, achieved throughput, completion latency, +and RSS together. Increase the budget only when the downstream service can +drain it; larger buffers can otherwise just extend queues and latency. The +eight-hour test also showed RSS growth, so it does not establish long-term memory +stability or a universally safe configuration. + +For callback writes, `writer_buffer_wait_timeout_ms` (client.writer.buffer.wait-timeout) +provides one shared budget for callback-capacity and buffer-backpressure waits. For `Wait()` and +fire-and-forget writes, the same setting governs the buffer wait. Set it to suit +upstream latency and overload handling, rather than assuming it bounds ACKs, +retries, or callback duration. Lowering it rejects sooner; it does +not cancel accepted writes. Increasing callback capacity does not increase Rust +buffer space, and increasing Rust buffer space does not prevent slow callbacks from +filling their operation limit. The default `writer_buffer_wait_timeout_ms = UINT64_MAX` +permits an unbounded wait; configure a finite value when callback submission or the +non-callback buffer wait needs to stop waiting and handle overload. + +### Execution and lifecycle + +Implement the callback's application logic; the SDK supplies the execution +threads. There is no need to create a waiting thread, call `Wait()`, or poll for +callback delivery. + +The SDK takes ownership of the callback and its captures. Callbacks run on a single +shared background worker in dispatch order, and may start before the +submitting call returns. Keep callbacks short; synchronize access to shared state and +keep captured references valid until the callback finishes. A callback that blocks +stalls the worker for every writer, so do not wait for other callbacks from inside one. +Callback overloads do not make writers safe for concurrent +access: serialize access if both the caller and a callback use the same writer. +Prefer capturing `std::shared_ptr` by value when sharing +application state. Keep the connection alive until outstanding operations +complete. Exceptions thrown by callbacks are caught and reported to stderr; +they do not change the write outcome or cause the callback to be invoked again. + +A callback should update thread-safe completion state and return promptly. +Capture an operation identifier by value so failures can be associated with +their input; `Result` does not contain the original row. If recovery is needed, +retain the payload or a reference to a durable source until its outcome has been +handled. The example above only logs failures; it is not a recovery implementation. + +For expensive processing or retries, hand off to an application event loop or +worker through a bounded, nonblocking mechanism. Define what happens when that +queue is full: preserve the failed operation and stop or backpressure new +submissions rather than silently dropping it or blocking SDK callback workers. +Do not loop on `Append`, sleep for retry backoff, or call `Flush()` inside a +callback. Schedule retries outside the callback with exclusive writer access. +They do not need to wait for `Flush()`, and remain subject to the duplicate risks +described above. A separate application thread is optional if an existing +submission loop can handle the handoff. + +Callbacks register directly with their internal write batch, rather than +creating an asynchronous ACK-waiting task for each row. When a batch completes, +its registered callbacks are dispatched together to the single process-wide +worker. Every callback still runs individually. Late registrations join the batch's +dispatch queue without overtaking its callbacks already awaiting dispatch. +Both Arrow log and KV +write batches use this path. An `AppendArrowBatch` spanning multiple internal +batches aggregates their results into the operation's single callback. + +The completion queue remains internally unbounded; per-writer admission limits +outstanding callback operations rather than dropping results from this queue. +This is not a byte or process-wide memory limit: capture sizes, batch sizes, the +number of writers, and application-owned retry queues need separate controls. +An aggregate ArrowBatch callback may report an error while later internal batches +are still pending, so its capacity slot does not bound all underlying batch memory. +Do not wait for another callback from within a callback, since the single callback +worker would be occupied. A callback submission from within any SDK write +callback fails immediately if its target writer's capacity is full, regardless +of `client.writer.buffer.wait-timeout`, to avoid blocking the worker on its own capacity. +This applies across writers as well. It does not +remove Rust buffer waits or make arbitrary blocking SDK calls deadlock-free. +Exclusive writer access is still required. The SDK initializes its callback worker +before accepting a callback write. Initialization failure returns a synchronous +client error without submitting data or registering a callback. It does not fall +back to a parallel pool; initialization failure remains in effect for the process. +An unexpected disconnection of the process-lifetime worker is a fatal internal +error, not a policy for discarding or reordering notifications. + +For shutdown, stop and join submitting threads, then call `Flush()` outside a +callback. It first runs the Rust write flush and, if that succeeds, blocks until +this writer's pending callbacks and captures finish, acting as a barrier. A callback +that never returns hangs it. +Check individual callback results as well: successful flushing is not a summary +that every submitted operation succeeded. + +If the write flush fails, `Flush()` returns that error and callbacks may still be +pending; do not release their referenced state. +`Flush()` does not wait for work handed to application workers or retry queues; +those need their own shutdown handling. Calling `Flush()` inside any SDK write +callback returns a client error before flushing, including calls on another writer. + +### Compatibility and operational limits + +- Existing fire-and-forget and `WriteResult::Wait()` overloads retain their + result semantics and do not consume callback capacity. Rust callers can still + `.await` a `WriteResultFuture`. Callback overloads share + `client.writer.buffer.wait-timeout` across capacity and buffer waits; + existing callback callers may block up to that timeout or receive a capacity or + buffer error. + Completion follows the configured acknowledgment policy; a callback does not + add a stronger durability guarantee or change retries, request ordering, + wire formats, or storage formats. +- The shared Rust completion path also changes for callers that do not register + callbacks: results are stored behind an `Arc`, callback registration state is + added per batch, and waiters are notified after releasing the result lock. + This changes allocation and scheduling costs, not the reported write outcome. + The boxed waiting future is now allocated on first poll rather than at + construction. Callback workers are initialized only when callbacks are used. +- For an `AppendArrowBatch` spanning multiple internal batches, success requires + all their results to succeed. As with `Wait()`, errors are selected in internal + handle order, not completion order; an error may be reported while later + batches remain pending. This is not an atomic multi-bucket write, and an error + does not imply that no rows were written. A submission error can also follow + partial acceptance of a multi-bucket batch; in that case no callback is + registered, so handle the returned error and do not assume an all-or-nothing retry. +- The single-worker execution and whole-batch job dispatch are implementation details, + not latency guarantees. A slow callback delays the callbacks queued behind it, and + slow callbacks from one connection can delay another connection. The worker is + initialized before callback writes are accepted; initialization failure rejects the + submission rather than switching to an unordered fallback. +- Callback delivery is in memory only. There is no end-to-end callback deadline + or durable recovery of pending notifications. + Connection or writer destruction is not a callback-drain barrier; process exit, + crashes, or fatal resource exhaustion can prevent pending callbacks from running. + The process-wide executor is not automatically drained at exit. +- Application tracking must allow callbacks before submission returns and + exclude rejected submissions, which have no callback. Follow the shutdown + sequence above before releasing callback state or the connection; an + application-side wait timeout does not cancel the write or its callback. +- C++ exceptions thrown by user callbacks are contained. If copying error text + fails, the callback still receives the error code, but the message may be empty. + Allocation failures while constructing the callback before submission can still + throw a C++ exception rather than return a `Result`. + +The new overloads keep ordinary existing calls source-compatible. Code that +selects an overload by taking a member-function address should specify the +intended function type explicitly. + ## `Lookuper` | Method | Description |