MINIFICPP-2740 Rust bindings based on stable C API by martinzink · Pull Request #2186 · apache/nifi-minifi-cpp

martinzink · 2026-05-28T12:21:06Z

Rebased onto #2205

Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.

In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:

For all changes:

Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically main)?
Is your initial contribution a single, squashed commit?

For code changes:

If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
If applicable, have you updated the LICENSE file?
If applicable, have you updated the NOTICE file?

For documentation related changes:

Have you ensured that format looks appropriate for the output in which it is rendered?

Note:

Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.

Copilot

Pull request overview

This PR introduces a new Rust workspace (minifi_rust/) that provides Rust bindings and an idiomatic safe API on top of the MiNiFi stable C API, along with a sample Rust extension and CI integration (unit tests + Behave-based Docker integration tests).

Changes:

Adds minifi_native_sys (bindgen-generated raw FFI) and minifi_native (safe Rust API + C-FFI glue + mocks/macros) crates.
Adds minifi_rs_playground example extension (processors/controller services) with unit tests and Behave integration tests.
Integrates Rust into the build/CI pipeline (CMake/Corrosion, Docker build scripts, GitHub Actions jobs/artifacts).

Reviewed changes

Copilot reviewed 127 out of 128 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
minifi_rust/rustfmt.toml	Rust formatting configuration for the workspace.
minifi_rust/README.md	Documents the Rust workspace architecture and extension authoring/deployment.
minifi_rust/minifi_rs_behave/src/main.rs	Rust test runner for Behave integration tests and optional Linux build trigger.
minifi_rust/minifi_rs_behave/linux_build.sh	Docker buildx wrapper to build Linux `.so` artifacts with SDK injection.
minifi_rust/minifi_rs_behave/debian.dockerfile	Debian-based container build for producing the Rust extension `.so`.
minifi_rust/minifi_rs_behave/Cargo.toml	Cargo manifest for the Behave runner crate.
minifi_rust/minifi_rs_behave/build.rs	Build script to expose Behave framework path to the runner.
minifi_rust/minifi_rs_behave/alpine.dockerfile	Alpine-based container build for producing the Rust extension `.so`.
minifi_rust/minifi_native/src/mock/mock_process_session.rs	Mock `ProcessSession` implementation for unit testing processors.
minifi_rust/minifi_native/src/mock/mock_process_context.rs	Mock `ProcessContext`/property map/controller service lookup for unit tests.
minifi_rust/minifi_native/src/mock/mock_logger.rs	Mock/stdout logger implementations + macro laziness test.
minifi_rust/minifi_native/src/mock/mock_flow_file.rs	Mock `FlowFile` implementation used by unit tests.
minifi_rust/minifi_native/src/mock/mock_controller_service_context.rs	Mock controller service context implementing `GetProperty`.
minifi_rust/minifi_native/src/mock.rs	Exposes mock types from the `minifi_native` crate.
minifi_rust/minifi_native/src/lib.rs	Crate root exports + extension registration macro + API version symbol.
minifi_rust/minifi_native/src/c_ffi/c_ffi_streams.rs	C-FFI input/output stream adapters implementing `BufRead`/`Write`.
minifi_rust/minifi_native/src/c_ffi/c_ffi_relationship.rs	Converts Rust relationship definitions into C representations.
minifi_rust/minifi_native/src/c_ffi/c_ffi_property.rs	Converts Rust property definitions + validators into C representations.
minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_list.rs	Holds processor definitions and exposes a C array pointer/count.
minifi_rust/minifi_native/src/c_ffi/c_ffi_process_context.rs	C-FFI `ProcessContext` implementation (properties/controller services).
minifi_rust/minifi_native/src/c_ffi/c_ffi_primitives.rs	Shared C-FFI primitives (string views, conversions, enums).
minifi_rust/minifi_native/src/c_ffi/c_ffi_output_attribute.rs	Converts output attributes into C representations with stable backing storage.
minifi_rust/minifi_native/src/c_ffi/c_ffi_logger.rs	C-FFI logger adapter delegating to MiNiFi’s logging callbacks.
minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs	C-FFI flow file wrapper type.
minifi_rust/minifi_native/src/c_ffi/c_ffi_controller_service_list.rs	Holds controller service definitions and exposes a C array pointer/count.
minifi_rust/minifi_native/src/c_ffi/c_ffi_controller_service_definition.rs	C-FFI controller service definition + callbacks wiring.
minifi_rust/minifi_native/src/c_ffi/c_ffi_controller_service_context.rs	C-FFI controller service context property retrieval.
minifi_rust/minifi_native/src/c_ffi.rs	Module wiring/re-exports for the C-FFI layer.
minifi_rust/minifi_native/src/api/relationship.rs	Defines the `Relationship` API type.
minifi_rust/minifi_native/src/api/raw_processor.rs	Defines raw processor traits + threading model markers.
minifi_rust/minifi_native/src/api/raw_controller_service.rs	Defines raw controller service trait.
minifi_rust/minifi_native/src/api/property.rs	Defines property schema + typed getters/validators.
minifi_rust/minifi_native/src/api/processor.rs	Defines scheduling/metrics/advanced features + processor wrapper struct.
minifi_rust/minifi_native/src/api/processor_wrappers/utils/flow_file_content.rs	Defines content representation (buffer/stream) for wrapper APIs.
minifi_rust/minifi_native/src/api/processor_wrappers/utils/context_session_flowfile_bundle.rs	Bundles context/session/flowfile to unify property/attribute access.
minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs	Exposes utility modules used by wrapper APIs.
minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs	Higher-level “transform” wrapper API for buffer/stream replacements.
minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs	Higher-level streaming transform wrapper API.
minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs	Higher-level source/generator wrapper API.
minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs	Wrapper for “complex” processors using trigger traits.
minifi_rust/minifi_native/src/api/processor_wrappers.rs	Module wiring for processor wrapper APIs.
minifi_rust/minifi_native/src/api/process_session.rs	Defines `ProcessSession`/stream traits + IO state.
minifi_rust/minifi_native/src/api/process_context.rs	Defines `ProcessContext` + typed property getters.
minifi_rust/minifi_native/src/api/logger.rs	Defines log levels, `Logger` trait, and logging macros.
minifi_rust/minifi_native/src/api/flow_file.rs	Defines the `FlowFile` marker trait.
minifi_rust/minifi_native/src/api/errors.rs	Defines `MinifiError` and conversions/display behavior.
minifi_rust/minifi_native/src/api/controller_service.rs	Defines controller service wrapper + enable behavior.
minifi_rust/minifi_native/src/api/component_definition_traits.rs	Defines processor/controller-service definition traits and IDs.
minifi_rust/minifi_native/src/api/attribute.rs	Defines output attributes and attribute getter trait.
minifi_rust/minifi_native/src/api.rs	API module wiring and public re-exports.
minifi_rust/minifi_native/Cargo.toml	Cargo manifest for the safe Rust API crate.
minifi_rust/minifi_native/.gitignore	Rust crate ignore rules.
minifi_rust/minifi_native_sys/src/lib.rs	Raw bindgen include wrapper for the generated FFI bindings.
minifi_rust/minifi_native_sys/Cargo.toml	Cargo manifest for the raw FFI crate and build dependencies.
minifi_rust/minifi_native_sys/build.rs	Build script to locate/download SDK, generate bindings, and emit metadata.
minifi_rust/minifi_native_sys/.gitignore	Rust crate ignore rules.
minifi_rust/minifi_native_macros/src/lib.rs	Proc-macros for component IDs, default metrics, and feature defaults.
minifi_rust/minifi_native_macros/Cargo.toml	Cargo manifest for the proc-macro crate.
minifi_rust/generate_docs/generate_docs.sh	Script to generate docs via containerized MiNiFi run.
minifi_rust/generate_docs/generate_docs.dockerfile	Dockerfile for docs generation against a MiNiFi container image.
minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/unix_only_properties.rs	PutFile Unix-only properties (permissions).
minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/tests.rs	Unit tests for the PutFile processor.
minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/relationships.rs	Relationship definitions for PutFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/properties.rs	Property definitions for PutFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs	ProcessorDefinition wiring for PutFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs	PutFile processor implementation using the Rust API.
minifi_rust/extensions/minifi_rs_playground/src/processors/mod.rs	Processor module wiring for the sample extension.
minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/tests.rs	Unit tests for controller-service usage processor.
minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/relationships.rs	Relationship definitions for LoremIpsumCSUser.
minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/properties.rs	Property definitions for LoremIpsumCSUser.
minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs	ProcessorDefinition wiring for LoremIpsumCSUser.
minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs	FlowFile source processor exercising controller services + content modes.
minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/tests.rs	Unit tests for LogAttribute processor behavior.
minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/relationships.rs	Relationship definitions for LogAttribute.
minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/properties.rs	Property definitions for LogAttribute.
minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs	ProcessorDefinition wiring for LogAttribute.
minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs	LogAttribute processor implementation using the Rust API.
minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs	Unit tests for failure/panic handling paths.
minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/relationships.rs	Relationship definitions for KamikazeProcessor.
minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/properties.rs	Property definitions for KamikazeProcessor.
minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs	ProcessorDefinition wiring for KamikazeProcessor.
minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs	Kamikaze test processor implementation.
minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs	Unit tests for GetFile processor.
minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/relationships.rs	Relationship definitions for GetFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/properties.rs	Property definitions for GetFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs	ProcessorDefinition wiring for GetFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/output_attributes.rs	Output attribute definitions for GetFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/tests.rs	Unit tests for GenerateFlowFile behavior.
minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/relationships.rs	Relationship definitions for GenerateFlowFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/properties.rs	Property definitions for GenerateFlowFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs	ProcessorDefinition wiring for GenerateFlowFile.
minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs	GenerateFlowFile processor implementation.
minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs	Streaming transform example processor.
minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs	Lazy logging test processor.
minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs	Unit tests for streaming transform cancellation/behavior.
minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/relationships.rs	Relationship definitions for AsciifyGerman.
minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs	ProcessorDefinition wiring for AsciifyGerman.
minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs	Streaming transform processor implementation.
minifi_rust/extensions/minifi_rs_playground/src/lib.rs	Declares and registers processors/controller services via the macro.
minifi_rust/extensions/minifi_rs_playground/src/controller_services/mod.rs	Controller service module wiring.
minifi_rust/extensions/minifi_rs_playground/src/controller_services/lorem_ipsum_controller_service/properties.rs	Property definitions for the lorem ipsum controller service.
minifi_rust/extensions/minifi_rs_playground/src/controller_services/lorem_ipsum_controller_service.rs	Controller service implementation generating lorem ipsum.
minifi_rust/extensions/minifi_rs_playground/src/controller_services/dummy_controller_service.rs	Dummy controller service for negative test cases.
minifi_rust/extensions/minifi_rs_playground/features/streaming.feature	Behave scenarios for streaming transforms.
minifi_rust/extensions/minifi_rs_playground/features/steps/steps.py	Custom Behave step implementations for new scenarios.
minifi_rust/extensions/minifi_rs_playground/features/metrics.feature	Behave scenarios validating metrics logging.
minifi_rust/extensions/minifi_rs_playground/features/lazy-logging.feature	Behave scenarios validating lazy logging behavior.
minifi_rust/extensions/minifi_rs_playground/features/error-handling.feature	Behave scenarios for error propagation and crash behavior.
minifi_rust/extensions/minifi_rs_playground/features/environment.py	Behave environment hooks to build/attach the Rust extension into containers.
minifi_rust/extensions/minifi_rs_playground/features/basic.feature	Basic Behave scenarios verifying extension loading and simple flows.
minifi_rust/extensions/minifi_rs_playground/Cargo.toml	Cargo manifest for the sample extension crate.
minifi_rust/extensions/minifi_rs_playground/.gitignore	Extension crate ignore rules.
minifi_rust/extensions/minifi_rs_playground/.cargo/config.toml	macOS linker flags for dynamic lookup in extensions.
minifi_rust/CMakeLists.txt	CMake integration for Rust via Corrosion and `cargo test` as a CTest.
minifi_rust/Cargo.toml	Workspace definition and profiles for panic behavior.
minifi_rust/.gitignore	Workspace ignore rules.
minifi_rust/.dockerignore	Docker ignore rules for the Rust workspace builds.
minifi_rust/.cargo/config.toml	Cargo aliases and env defaults for SDK/test execution.
docker/rockylinux/Dockerfile	Updates base image and installs rust tooling conditionally for Rust builds.
CMakeLists.txt	Adds `minifi_rust` subdirectory to the top-level build.
cmake/MiNiFiOptions.cmake	Adds `MINIFI_RUST` CMake option to control Rust integration.
.github/workflows/create-release-artifacts.yml	Builds Rust artifacts and uploads `.so` extension outputs.
.github/workflows/ci.yml	Adds Rust docker integration test job and Rust formatting check in linters.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+impl<S> GetControllerService for S
+where
+    S: ProcessContext,
+{
+    fn get_controller_service<Cs>(&self, property: &Property) -> Result<Option<&Cs>, MinifiError>
+    where
+        Cs: EnableControllerService + ComponentIdentifier + 'static,
+    {
+        self.get_controller_service(property)
+    }
+}


+    fn get_destination_path<Ctx>(context: &Ctx) -> Result<PathBuf, MinifiError>
+    where
+        Ctx: GetProperty + GetAttribute,
+    {
+        let directory = context
+            .get_property(&properties::DIRECTORY)?
+            .expect("required property");
+
+        let file_name = context
+            .get_attribute("filename")?
+            .unwrap_or("foo.txt".to_string()); // TODO fallback to UUID
+        Ok(PathBuf::from(directory + "/" + file_name.as_str()))
+    }


  linters:
-    name: "C++ lint + Shellcheck + Flake8"
+    name: "C++ lint + Shellcheck + Flake8 + Cargo check"
    runs-on: ubuntu-22.04-arm
    timeout-minutes: 15
    steps:
      - id: checkout
        uses: actions/checkout@v6

      - id: install_deps
        run: sudo apt update && sudo apt install -y flake8

      - id: cpp_lint
        name: C++ linter
        continue-on-error: true
        run: python3 thirdparty/google-styleguide/run_linter.py -q -i controller/ core-framework/ encrypt-config/ extension-framework/ extensions/ libminifi/ minifi-api/ minifi_main/

      - id: shellcheck
        name: Shellcheck
        continue-on-error: true
        run: ./run_shellcheck.sh .

      - id: flake8_check
        name: Flake8 check
        continue-on-error: true
        run: ./run_flake8.sh .

+      - id: cargo_check
+        name: Cargo check
+        continue-on-error: true
+        run: cargo fmt --check
+        working-directory: minifi_rust
+


martinzink · 2026-05-28T12:26:33Z

+from minifi_behave.core.minifi_test_context import MinifiTestContext
+
+
+@when("the MiNiFi instance is started without assertions")


We should probably move these into the framework later, but I didnt want to touch anything outside of minifi_rust dir

martinzink · 2026-05-28T12:28:54Z

@@ -0,0 +1,13 @@
+use minifi_native::{Property, StandardPropertyValidator};
+
+pub(crate) const LENGTH: Property = Property {


We should discuss what kind of style we want to follow. Single file per processor or splitting the properties, relationships, processor definitions into the mods folder

I think I'd keep all parts of a processor in the same file, because it's a single logical unit of code in my view.

Im a bit torn, on one hand u are right on the other it might be benefitial to move the processor interface out to a separate file (so relationships, properties, processor definition) so does large heavily boilerplated parts dont pollute the actual code.

I think we can change these in an upcoming PR after we came to an agreement.

When we are using the multiple-file style, can we put the processor definition at the parent level and move the implementation to the subdirectory? That would make more sense to me, since the processor definition file is similar to a table of contents, and it also contains the description of the processor.

martinzink · 2026-05-28T12:30:33Z

@@ -0,0 +1,2 @@
+pub(crate) mod dummy_controller_service;
+pub(crate) mod lorem_ipsum_controller_service;


test controller services, (multiple so we can test setting and quering the wrong one)

martinzink · 2026-05-28T12:31:20Z

@@ -0,0 +1,67 @@
+use crate::processors::asciify_german::relationships::FAILURE;


This is used to test streaming flow file transforms, with changing flowfiles sizes

I'd prefer to have this explanation in a code comment

martinzink · 2026-05-28T12:39:39Z

@@ -0,0 +1,20 @@
+[package]
+name = "minifi_rs_playground"


This extension contains reimplentations of already existing processors and test only processors. These are not intended for production usage, therefore they probably don't require that much scrutiny in the review process.

If for some reason we ever want to ship something from this extension than that processor will have to be extracted and rereviewed.

martinzink · 2026-06-02T07:31:52Z

+    Property,
+};
+
+pub struct ContextSessionFlowFileBundle<'a, PC, PS>


This allows a single "context" for the transform (higher API) processors while maintaning the functionaility spread across the context and session.

martinzink · 2026-06-02T07:33:38Z

@@ -0,0 +1,27 @@
+use std::fmt::Formatter;
+
+pub enum Content<'a> {


This makes the transform return value to be more usable, it can either return the content in a buffer or somewhere where it can be read from (if we are concerned with memory usage)

Concurrent -> MultiThreaded, Exclusive -> SingleThreaded

szaszm

halfway

szaszm · 2026-07-20T11:00:32Z

@@ -0,0 +1,13 @@
+use minifi_native::{Property, StandardPropertyValidator};
+
+pub(crate) const LENGTH: Property = Property {


I think I'd keep all parts of a processor in the same file, because it's a single logical unit of code in my view.

szaszm · 2026-07-20T16:16:00Z

@@ -0,0 +1,67 @@
+use crate::processors::asciify_german::relationships::FAILURE;


I'd prefer to have this explanation in a code comment

lordgamez

LGTM only some minor nitpicks

szaszm · 2026-07-21T16:57:48Z

+minifi_native::declare_minifi_extension!(
+processors: [


I think we should indent the contents within declare_minifi_extension!()

the one place where auto format doesnt apply 😅 fixed here indent minifi_rs_playground/src/lib.rs macro

szaszm

almost done. One more ask: please consider moving some of your explanatory github comments into code comments, I think they would be more useful in the future if they were in the source files.

fgerlits · 2026-07-20T10:23:20Z

+
+@when("the MiNiFi instance is started without assertions")
+def minifi_starts_wo_assertions(context: MinifiTestContext):
+    context.get_or_create_default_minifi_container().deploy(context)


how is this different from the standard "the MiNiFi instance starts up" step?

fgerlits · 2026-07-23T08:06:29Z

+    pub supports_expr_lang: bool,
+    pub default_value: Option<&'static str>,
+    pub validator: StandardPropertyValidator,
+    pub allowed_values: &'static [&'static str],


Should allowed_values be an Option, too? I think it would make more sense to use None to mean "no restrictions", because an empty slice looks like "no values are allowed".

I don't know if the type system can represent "optional slice of strings, but if it is Some, then it is non-empty", but if it can, that would be really nice.

fgerlits · 2026-07-23T08:41:10Z

+
+pub(crate) const SUCCESS: Relationship = Relationship {
+    name: "success",
+    description: "FlowFiles are transferred here after logging",


nitpicking:

Suggested change

description: "FlowFiles are transferred here after logging",

description: "The created FlowFiles are transferred here",

fgerlits · 2026-07-23T09:01:34Z

@@ -0,0 +1,13 @@
+use minifi_native::{Property, StandardPropertyValidator};
+
+pub(crate) const LENGTH: Property = Property {


When we are using the multiple-file style, can we put the processor definition at the parent level and move the implementation to the subdirectory? That would make more sense to me, since the processor definition file is similar to a table of contents, and it also contains the description of the processor.

fgerlits · 2026-07-23T12:09:48Z

+
+// minifi-sys/src/lib.rs


stale comment?

Suggested change

// minifi-sys/src/lib.rs

fgerlits · 2026-07-23T12:15:06Z

+
+    let bindings = bindgen::Builder::default()
+        .header(sdk.header_path.to_str().unwrap())
+        .clang_arg("-std=c2x")


Can this "-std=c2x" come from the Cargo.toml file? It looks like something which will need regular updates, and it will be easy to forget if it is hidden here.

szaszm

last batch

szaszm · 2026-07-23T12:04:58Z

+            match processor.trigger(&mut context, &mut session) {
+                Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
+                Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
+                Err(error_code) => error_code.to_status(),


Errors are logged in the MultiThreaded case, but not here. Why?

szaszm · 2026-07-23T12:22:02Z

+    ptr: *mut minifi_input_stream,
+    buffer: [u8; 8192],
+    pos: usize,
+    cap: usize,


What does cap mean here?

szaszm · 2026-07-23T13:00:00Z

+                    loop {
+                        let n = match (state.callback)(state.buffer) {
+                            Ok(Some(n)) => n,
+                            Ok(None) => break, // EOF
+                            Err(err) => {
+                                state.error = Some(err);
+                                return minifi_io_status_MINIFI_IO_ERROR;
+                            }
+                        };
+
+                        let written = minifi_output_stream_write(
+                            output_stream,
+                            state.buffer.as_ptr() as *const c_char,
+                            n,
+                        );
+
+                        if written < 0 {
+                            return minifi_io_status_MINIFI_IO_ERROR;
+                        }
+                        overall_writes += written;
+                    }


Partial writes are not handled, each batch is only tried once, even if only a part of them is successfully written to the output stream. Copilot pointed my attention here, original output:

Short writes can silently truncate content
write_in_batches writes each produced buffer exactly once (lines 105–114), and write does the same (lines 308–312). The C API returns a byte count, so either call must retry until the entire supplied slice is written; otherwise any partial write drops the unwritten suffix.
View lines 95–116 · View lines 295–329

szaszm · 2026-07-23T13:32:14Z

+        attr_value
+    }
+
+    fn on_attributes<F: FnMut(&str, &str)>(


I think on_attributes as a name is a bit ambiguous, "on" can mean many things. I'd consider alternatives such as for_each_attribute or observe_all_attributes

szaszm · 2026-07-23T13:37:23Z

+            ) -> i64 {
+                unsafe {
+                    let result_target = &mut *(user_ctx as *mut Option<&[u8]>);
+                    if result_target.is_none() {
+                        return minifi_io_status_MINIFI_IO_ERROR;
+                    }
+
+                    minifi_output_stream_write(
+                        output_stream,
+                        result_target.unwrap().as_ptr() as *const c_char,
+                        result_target.unwrap().len(),
+                    )
+                }
+            }


partial writes may be a problem here too. I believe a new session write will start a new content blob, so the caller has no way of retrying a partial write, if it can happen at all. (which depends on the specific stream implementations)

szaszm · 2026-07-23T14:23:02Z

+                    if bytes_read < 0 {
+                        *result_target = None;
+                        return bytes_read;
+                    }
+
+                    buffer.set_len(bytes_read as usize);
+
+                    *result_target = Some(buffer);


What happens if bytes_read > 0 && bytes_read < stream_size?

szaszm · 2026-07-23T14:24:01Z

+            minifi_process_session_read(
+                self.ptr,
+                flow_file.get_ptr(),
+                Some(cb),
+                &mut output as *mut _ as *mut c_void,
+            );


The return value is discarded, is this on purpose?

szaszm · 2026-07-23T14:26:07Z

+            minifi_process_session_read(
+                self.ptr,
+                flow_file.get_ptr(),
+                Some(read_cb::<F, R>),
+                &mut ctx as *mut _ as *mut c_void,
+            );
+
+            ctx.result
+                .expect("Agent returned with success, so ctx.result should be set")


The status code is ignored, so it seems to me that it's possible that this will panic on read failure.

related AI review output:

read_stream ignores the C API status and may panic on normal FFI failure
It discards the return value from minifi_process_session_read, then calls expect if its callback did not run (lines 497–505). The C API can return MINIFI_STATUS_UNKNOWN_ERROR without invoking the callback, so this turns an FFI error into a Rust panic. It should inspect the returned status and return MinifiError::StatusError; likewise, read() currently ignores that status and reports failure only as None.
View lines 463–506 · C API behavior

szaszm · 2026-07-23T14:28:51Z

+                    if let Some(cb) = ctx.callback.take() {
+                        ctx.result = Some(cb(&mut reader));
+                    }
+
+                    reader.total_bytes_read() as i64


shouldn't the C callback propagate rust callback errors to minifi?

szaszm · 2026-07-23T14:32:24Z

+    fn read_in_batches<F>(
+        &self,
+        flow_file: &Self::FlowFile,
+        batch_size: usize,
+        process_batch: F,
+    ) -> Result<(), MinifiError>
+    where
+        F: FnMut(&[u8]) -> Result<(), MinifiError>,
+    {


AI review flagged that if batch_size is zero, then this can get stuck in an infinite loop, and it seems to be the case to me too. Maybe we could change batch_size to NonZeroUsize to prevent this.

AI output:

read_in_batches can loop forever with batch_size == 0
At lines 541–566, read_size becomes zero, the read returns zero, and remaining_size never decreases. Reject a zero batch size up front (and treat a zero-byte read while data remains as an I/O error).
View lines 509–568

martinzink requested a review from Copilot May 28, 2026 12:21

Copilot started reviewing on behalf of martinzink May 28, 2026 12:21 View session

Copilot AI reviewed May 28, 2026

View reviewed changes

martinzink commented May 28, 2026

View reviewed changes

martinzink force-pushed the minifi_rust branch from 3ddaa25 to c71d822 Compare June 2, 2026 07:27