reifydb_sub_flow/operator/
extern_rust.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	collections::HashMap,
6	path::{Path, PathBuf},
7	sync::OnceLock,
8};
9
10use libloading::Symbol;
11use reifydb_core::interface::catalog::flow::OperatorId;
12use reifydb_extension::loader::extern_load::ExternLoad;
13use reifydb_flow::operator::BoxedHostOperator;
14use reifydb_runtime::sync::rwlock::RwLock;
15use reifydb_value::{Result, config::Config, error::Error, value::constraint::TypeConstraint};
16
17use crate::error::ExternOperatorError;
18
19pub const EXTERN_RUST_OPERATOR_MAGIC: u32 = 0x5244_424E;
20
21pub const EXTERN_RUST_ABI_TAG: u32 = 0x0308;
22
23pub type ExternRustOperatorCreateFn = fn(OperatorId, &Config) -> Result<BoxedHostOperator>;
24
25pub struct ExternRustOperatorColumn {
26	pub name: String,
27	pub field_type: TypeConstraint,
28	pub description: String,
29}
30
31pub struct ExternRustOperatorDescriptor {
32	pub abi_tag: u32,
33	pub name: String,
34	pub version: String,
35	pub description: String,
36	pub capabilities: u32,
37	pub input_columns: Vec<ExternRustOperatorColumn>,
38	pub output_columns: Vec<ExternRustOperatorColumn>,
39}
40
41pub fn extern_rust_operator_magic() -> u32 {
42	EXTERN_RUST_OPERATOR_MAGIC
43}
44
45pub fn check_extern_rust_abi_tag(abi_tag: u32) -> Result<()> {
46	if abi_tag != EXTERN_RUST_ABI_TAG {
47		return Err(Error::from(ExternOperatorError::AbiTagMismatch {
48			plugin: abi_tag,
49			host: EXTERN_RUST_ABI_TAG,
50		}));
51	}
52	Ok(())
53}
54
55pub struct LoadedExternRustOperatorInfo {
56	pub operator: String,
57	pub library_path: PathBuf,
58	pub abi_tag: u32,
59	pub version: String,
60	pub description: String,
61	pub input_columns: Vec<ExternRustOperatorColumn>,
62	pub output_columns: Vec<ExternRustOperatorColumn>,
63	pub capabilities: u32,
64}
65
66static GLOBAL_EXTERN_RUST_OPERATOR_LOADER: OnceLock<RwLock<ExternRustOperatorLoader>> = OnceLock::new();
67
68pub fn extern_rust_operator_loader() -> &'static RwLock<ExternRustOperatorLoader> {
69	GLOBAL_EXTERN_RUST_OPERATOR_LOADER.get_or_init(|| RwLock::new(ExternRustOperatorLoader::new()))
70}
71
72pub struct ExternRustOperatorLoader {
73	cache: ExternLoad,
74	operator_paths: HashMap<String, PathBuf>,
75}
76
77impl ExternRustOperatorLoader {
78	fn new() -> Self {
79		Self {
80			cache: ExternLoad::new(),
81			operator_paths: HashMap::new(),
82		}
83	}
84
85	fn load_library(&mut self, path: &Path) -> Result<bool> {
86		self.cache
87			.check_magic(path, b"reifydb_extern_rust_operator_magic\0", EXTERN_RUST_OPERATOR_MAGIC)
88			.map_err(|_e| {
89				Error::from(ExternOperatorError::LibraryNotLoaded {
90					path: path.display().to_string(),
91				})
92			})
93	}
94
95	fn descriptor(&self, path: &Path) -> Result<ExternRustOperatorDescriptor> {
96		let library = self.cache.get(path).ok_or_else(|| {
97			Error::from(ExternOperatorError::LibraryNotLoaded {
98				path: path.display().to_string(),
99			})
100		})?;
101
102		// SAFETY: load_library accepted this path only after the extern-Rust operator magic symbol
103		// matched, so the object was built against this crate and declares the descriptor symbol with
104		// this signature; Symbol borrows library, which stays loaded for the call.
105		let descriptor = unsafe {
106			let get_descriptor: Symbol<fn() -> ExternRustOperatorDescriptor> =
107				library.get(b"reifydb_extern_rust_operator_descriptor\0").map_err(|e| {
108					Error::from(ExternOperatorError::SymbolNotFound {
109						symbol: "reifydb_extern_rust_operator_descriptor",
110						cause: e.to_string(),
111					})
112				})?;
113			get_descriptor()
114		};
115
116		check_extern_rust_abi_tag(descriptor.abi_tag)?;
117
118		Ok(descriptor)
119	}
120
121	pub fn register_operator(&mut self, path: &Path) -> Result<Option<LoadedExternRustOperatorInfo>> {
122		if !self.load_library(path)? {
123			return Ok(None);
124		}
125
126		let descriptor = self.descriptor(path)?;
127		self.operator_paths.insert(descriptor.name.clone(), path.to_path_buf());
128
129		Ok(Some(LoadedExternRustOperatorInfo {
130			operator: descriptor.name,
131			library_path: path.to_path_buf(),
132			abi_tag: descriptor.abi_tag,
133			version: descriptor.version,
134			description: descriptor.description,
135			input_columns: descriptor.input_columns,
136			output_columns: descriptor.output_columns,
137			capabilities: descriptor.capabilities,
138		}))
139	}
140
141	pub fn has_operator(&self, operator: &str) -> bool {
142		self.operator_paths.contains_key(operator)
143	}
144
145	pub fn create_operator_by_name(
146		&mut self,
147		operator: &str,
148		operator_id: OperatorId,
149		config: &Config,
150	) -> Result<BoxedHostOperator> {
151		let path = self
152			.operator_paths
153			.get(operator)
154			.ok_or_else(|| {
155				Error::from(ExternOperatorError::OperatorNotFound {
156					operator: operator.to_string(),
157				})
158			})?
159			.clone();
160
161		if !self.load_library(&path)? {
162			return Err(Error::from(ExternOperatorError::LibraryNotLoaded {
163				path: operator.to_string(),
164			}));
165		}
166
167		self.descriptor(&path)?;
168
169		let library = self.cache.library(&path).map_err(|_| {
170			Error::from(ExternOperatorError::LibraryNotLoaded {
171				path: operator.to_string(),
172			})
173		})?;
174		// SAFETY: load_library and descriptor accepted this path, so the object was built against this
175		// crate and declares the create symbol with this signature; the copied-out pointer is called
176		// before this method returns, while &mut self still holds the cache entry that keeps it mapped.
177		let create: ExternRustOperatorCreateFn = unsafe {
178			let create_symbol: Symbol<ExternRustOperatorCreateFn> =
179				library.get(b"reifydb_extern_rust_operator_create\0").map_err(|e| {
180					Error::from(ExternOperatorError::SymbolNotFound {
181						symbol: "reifydb_extern_rust_operator_create",
182						cause: e.to_string(),
183					})
184				})?;
185			*create_symbol
186		};
187
188		create(operator_id, config)
189	}
190}
191
192impl Default for ExternRustOperatorLoader {
193	fn default() -> Self {
194		Self::new()
195	}
196}
197
198#[cfg(test)]
199mod tests {
200	use reifydb_extension::operator::extern_c::loader::check_operator_abi_tag;
201	use reifydb_sdk::flow::operator::extern_c::wire::types::OPERATOR_ABI_TAG;
202
203	use super::{EXTERN_RUST_ABI_TAG, check_extern_rust_abi_tag};
204
205	#[test]
206	fn extern_rust_abi_tag_accepts_match_rejects_mismatch() {
207		// A mismatched tag must be refused, or an operator built against a different toolchain gets loaded.
208		assert!(check_extern_rust_abi_tag(EXTERN_RUST_ABI_TAG).is_ok());
209		assert!(check_extern_rust_abi_tag(EXTERN_RUST_ABI_TAG ^ 0x1).is_err());
210		assert!(check_extern_rust_abi_tag(0).is_err());
211	}
212
213	#[test]
214	fn extern_c_abi_tag_accepts_match_rejects_mismatch() {
215		assert!(check_operator_abi_tag(OPERATOR_ABI_TAG).is_ok());
216		assert!(check_operator_abi_tag(OPERATOR_ABI_TAG ^ 0x1).is_err());
217		assert!(check_operator_abi_tag(0).is_err());
218	}
219
220	#[test]
221	fn extern_rust_and_extern_c_tags_do_not_accept_each_other() {
222		// The two tags must reject each other, or an extern-Rust `.so` validates against the extern-C check.
223		assert_ne!(EXTERN_RUST_ABI_TAG, OPERATOR_ABI_TAG);
224		assert!(check_extern_rust_abi_tag(OPERATOR_ABI_TAG).is_err());
225		assert!(check_operator_abi_tag(EXTERN_RUST_ABI_TAG).is_err());
226	}
227}
reifydb_sub_flow/operator/extern_rust.rs

reifydb_sub_flow/operator/
extern_rust.rs
