picodata_plugin/plugin/
interface.rs

1use crate::background;
2use crate::error_code::ErrorCode;
3use crate::util::FfiSafeStr;
4pub use abi_stable;
5use abi_stable::pmr::{RErr, RResult, RSlice};
6use abi_stable::std_types::{RBox, RHashMap, ROk, ROption, RString, RVec};
7use abi_stable::{sabi_trait, RTuple, StableAbi};
8use serde::de::DeserializeOwned;
9use smol_str::SmolStr;
10use std::collections::HashMap;
11use std::error::Error;
12use std::fmt::Display;
13use std::time::Duration;
14use tarantool::error::TarantoolErrorCode;
15use tarantool::error::{BoxError, IntoBoxError};
16
17/// Context of current instance. Produced by picodata.
18#[repr(C)]
19#[derive(StableAbi, Debug)]
20pub struct PicoContext {
21    is_master: bool,
22    pub plugin_name: FfiSafeStr,
23    pub service_name: FfiSafeStr,
24    pub plugin_version: FfiSafeStr,
25    #[sabi(last_prefix_field)]
26    is_raft_leader: bool,
27}
28
29impl PicoContext {
30    #[inline]
31    pub fn new(is_master: bool) -> PicoContext {
32        Self {
33            is_master,
34            plugin_name: "<unset>".into(),
35            service_name: "<unset>".into(),
36            plugin_version: "<unset>".into(),
37            is_raft_leader: false,
38        }
39    }
40
41    #[inline]
42    pub fn with_is_raft_leader(is_master: bool, is_raft_leader: bool) -> PicoContext {
43        Self {
44            is_master,
45            plugin_name: "<unset>".into(),
46            service_name: "<unset>".into(),
47            plugin_version: "<unset>".into(),
48            is_raft_leader,
49        }
50    }
51
52    /// # Safety
53    ///
54    /// Note: this is for internal use only. Plugin developers should never
55    /// be copying pico context.
56    #[inline]
57    pub unsafe fn clone(&self) -> Self {
58        Self {
59            is_master: self.is_master,
60            plugin_name: self.plugin_name,
61            service_name: self.service_name,
62            plugin_version: self.plugin_version,
63            is_raft_leader: self.is_raft_leader,
64        }
65    }
66
67    /// Return true if the current instance is a replicaset leader.
68    #[inline]
69    pub fn is_master(&self) -> bool {
70        self.is_master
71    }
72
73    /// Return true if the current instance is a raft leader.
74    #[inline]
75    pub fn is_raft_leader(&self) -> bool {
76        self.is_raft_leader
77    }
78
79    #[inline(always)]
80    pub fn register_metrics_callback(&self, callback: impl Fn() -> String) -> Result<(), BoxError> {
81        crate::metrics::register_metrics_handler(self, callback)
82    }
83
84    /// Add a new job to the execution.
85    /// Job work life cycle will be tied to the service life cycle;
86    /// this means that job will be canceled just before service is stopped.
87    ///
88    /// # Arguments
89    ///
90    /// * `job`: callback that will be executed in separated fiber.
91    ///   Note that it is your responsibility to organize job graceful shutdown, see a
92    ///   [`background::CancellationToken`] for details.
93    ///
94    /// # Examples
95    ///
96    /// ```no_run
97    /// use std::time::Duration;
98    /// use picodata_plugin::background::CancellationToken;
99    ///
100    /// # use picodata_plugin::plugin::interface::PicoContext;
101    /// # fn on_start(context: PicoContext) {
102    ///
103    /// // this job will print "hello" every second,
104    /// // and print "bye" after being canceled
105    /// fn hello_printer(cancel: CancellationToken) {
106    ///     while cancel.wait_timeout(Duration::from_secs(1)).is_err() {
107    ///         println!("hello!");
108    ///     }
109    ///     println!("job cancelled, bye!")
110    /// }
111    /// context.register_job(hello_printer).unwrap();
112    ///
113    /// # }
114    /// ```
115    #[track_caller]
116    #[inline(always)]
117    pub fn register_job<F>(&self, job: F) -> Result<(), BoxError>
118    where
119        F: FnOnce(background::CancellationToken) + 'static,
120    {
121        background::register_job(&self.make_service_id(), job)
122    }
123
124    /// Same as [`Self::register_job`] but caller may provide a special tag.
125    /// This tag may be used for manual job cancellation using [`Self::cancel_tagged_jobs`].
126    ///
127    /// # Arguments
128    ///
129    /// * `job`: callback that will be executed in separated fiber
130    /// * `tag`: tag, that will be related to a job, single tag may be related to the multiple jobs
131    #[inline(always)]
132    pub fn register_tagged_job<F>(&self, job: F, tag: &str) -> Result<(), BoxError>
133    where
134        F: FnOnce(background::CancellationToken) + 'static,
135    {
136        background::register_tagged_job(&self.make_service_id(), job, tag)
137    }
138
139    /// Cancel all jobs related to the given `tag`.
140    /// This function return after all related jobs will be gracefully shutdown or
141    /// after `timeout` duration.
142    ///
143    /// Returns error with code [`TarantoolErrorCode::Timeout`] in case some
144    /// jobs didn't finish within `timeout`.
145    ///
146    /// May also theoretically return error with code [`ErrorCode::NoSuchService`]
147    /// in case the service doesn't exist anymore (highly unlikely).
148    ///
149    /// See also [`Self::register_tagged_job`].
150    #[inline(always)]
151    pub fn cancel_tagged_jobs(&self, tag: &str, timeout: Duration) -> Result<(), BoxError> {
152        let res = background::cancel_jobs_by_tag(&self.make_service_id(), tag, timeout)?;
153        if res.n_timeouts != 0 {
154            #[rustfmt::skip]
155            return Err(BoxError::new(TarantoolErrorCode::Timeout, format!("some background jobs didn't finish in time (expected: {}, timed out: {})", res.n_total, res.n_timeouts)));
156        }
157
158        Ok(())
159    }
160
161    /// In case when jobs were canceled by `picodata` use this function for determine
162    /// a shutdown timeout - time duration that `picodata` uses to ensure that all
163    /// jobs gracefully end.
164    ///
165    /// By default, 5-second timeout are used.
166    #[inline(always)]
167    pub fn set_jobs_shutdown_timeout(&self, timeout: Duration) {
168        crate::background::set_jobs_shutdown_timeout(
169            self.plugin_name(),
170            self.service_name(),
171            self.plugin_version(),
172            timeout,
173        )
174    }
175
176    #[inline(always)]
177    pub fn make_service_id(&self) -> ServiceId {
178        ServiceId::new(
179            self.plugin_name(),
180            self.service_name(),
181            self.plugin_version(),
182        )
183    }
184
185    #[inline]
186    pub fn plugin_name(&self) -> &str {
187        // SAFETY: safe because lifetime is managed by borrow checker
188        unsafe { self.plugin_name.as_str() }
189    }
190
191    #[inline]
192    pub fn service_name(&self) -> &str {
193        // SAFETY: safe because lifetime is managed by borrow checker
194        unsafe { self.service_name.as_str() }
195    }
196
197    #[inline]
198    pub fn plugin_version(&self) -> &str {
199        // SAFETY: safe because lifetime is managed by borrow checker
200        unsafe { self.plugin_version.as_str() }
201    }
202}
203
204/// Unique service identifier.
205#[derive(Debug, PartialEq, Eq, Hash, Clone)]
206pub struct ServiceId {
207    pub plugin: SmolStr,
208    pub service: SmolStr,
209    pub version: SmolStr,
210}
211
212impl std::fmt::Display for ServiceId {
213    #[inline(always)]
214    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215        write!(f, "{}.{}:v{}", self.plugin, self.service, self.version,)
216    }
217}
218
219impl ServiceId {
220    #[inline(always)]
221    pub fn new(
222        plugin: impl Into<SmolStr>,
223        service: impl Into<SmolStr>,
224        version: impl Into<SmolStr>,
225    ) -> Self {
226        Self {
227            plugin: plugin.into(),
228            service: service.into(),
229            version: version.into(),
230        }
231    }
232
233    #[inline(always)]
234    pub fn plugin(&self) -> &str {
235        &self.plugin
236    }
237
238    #[inline(always)]
239    pub fn service(&self) -> &str {
240        &self.service
241    }
242
243    #[inline(always)]
244    pub fn version(&self) -> &str {
245        &self.version
246    }
247}
248
249// --------------------------- user interface ------------------------------------------------------
250
251/// Error type, return it from your callbacks.
252pub type ErrorBox = Box<dyn Error>;
253
254pub type CallbackResult<T> = Result<T, ErrorBox>;
255
256/// Service trait. Implement it in your code to create a service.
257pub trait Service {
258    /// Use this associated type to define configuration of your service.
259    type Config: DeserializeOwned;
260
261    /// Callback to handle service configuration change once instance receives it.
262    ///
263    /// # Idempotency
264    ///
265    /// **WARNING** This callback may be called several times in a row.
266    /// It is the responsibility of the plugin author to make this function idempotent.
267    ///
268    /// # Poison
269    ///
270    /// Return an error here to poison current instance.
271    /// This will not cancel reconfiguration process,
272    /// but will mark instance as unavailable for rpc messaging.
273    ///
274    /// Instance can be "healed"
275    /// if any of `on_replicaset_leader_change`, `on_config_change` or `on_cluster_leader_change`
276    /// callbacks in the future return `Ok`.
277    ///
278    /// # Arguments
279    ///
280    /// * `ctx`: instance context
281    /// * `new_config`: new configuration
282    /// * `old_config`: previous defined configuration
283    fn on_config_change(
284        &mut self,
285        ctx: &PicoContext,
286        new_config: Self::Config,
287        old_config: Self::Config,
288    ) -> CallbackResult<()> {
289        _ = ctx;
290        _ = new_config;
291        _ = old_config;
292        Ok(())
293    }
294
295    /// Called at service start on every instance.
296    ///
297    /// # Idempotency
298    ///
299    /// **WARNING** This callback may be called several times in a row (without
300    /// any calls to [`Self::on_stop`]). It is the responsibility of the plugin
301    /// author to make this function idempotent.
302    ///
303    /// An error returned here abort plugin load clusterwide thus forcing
304    /// `on_stop` callback execution on every instance.
305    ///
306    /// # Arguments
307    ///
308    /// * `context`: instance context
309    /// * `config`: initial configuration
310    fn on_start(&mut self, context: &PicoContext, config: Self::Config) -> CallbackResult<()> {
311        _ = context;
312        _ = config;
313        Ok(())
314    }
315
316    /// Called on instance shutdown, plugin removal or failure of the initial load.
317    /// Returned error will only be logged causing no effects on plugin lifecycle.
318    ///
319    /// # Idempotency
320    ///
321    /// **WARNING** This callback may be called several times in a row.
322    /// It is the responsibility of the plugin author to make this function idempotent.
323    ///
324    /// # Arguments
325    ///
326    /// * `context`: instance context
327    fn on_stop(&mut self, context: &PicoContext) -> CallbackResult<()> {
328        _ = context;
329        Ok(())
330    }
331
332    /// This method will be deprecated in the next major release in favor of
333    /// `on_replicaset_leader_change`. Only the name is changed, so you can easily migrate
334    /// to the new method.
335    #[deprecated(
336        since = "26.2.1",
337        note = "this method gets deprecated, in favor of on_replicaset_leader_change for better naming. Both callbacks will be called until the next major release"
338    )]
339    fn on_leader_change(&mut self, context: &PicoContext) -> CallbackResult<()> {
340        _ = context;
341        Ok(())
342    }
343
344    /// Called when the replicaset master is changed.
345    /// This callback will be called on two instances at max - the old replicaset
346    /// master (if it is accessible) and the new one.
347    ///
348    /// # Idempotency
349    ///
350    /// **WARNING** This callback may be called several times in a row.
351    /// It is the responsibility of the plugin author to make this function idempotent.
352    ///
353    /// # Poison
354    ///
355    /// Return an error here to poison current instance.
356    /// This will not cancel reconfiguration process,
357    /// but will mark instance as unavailable for rpc messaging.
358    ///
359    /// Instance can be "healed"
360    /// if any of `on_replicaset_leader_change`, `on_config_change` or `on_cluster_leader_change`
361    /// callbacks in the future return `Ok`.
362    ///
363    /// # Arguments
364    ///
365    /// * `context`: instance context
366    fn on_replicaset_leader_change(&mut self, context: &PicoContext) -> CallbackResult<()> {
367        _ = context;
368        Ok(())
369    }
370
371    /// Called when the raft leader of the cluster is changed
372    /// This callback will be called on two instances at max - the old cluster
373    /// leader (if it is accessible) and the new one.
374    ///
375    /// # Idempotency
376    ///
377    /// **WARNING** This callback may be called several times in a row.
378    /// It is the responsibility of the plugin author to make this function idempotent.
379    ///
380    /// # Poison
381    ///
382    /// Return an error here to poison current instance.
383    /// This will not cancel reconfiguration process,
384    /// but will mark instance as unavailable for rpc messaging.
385    ///
386    /// Instance can be "healed"
387    /// if any of `on_replicaset_leader_change`, `on_config_change` or `on_cluster_leader_change`
388    /// callbacks in the future return `Ok`.
389    ///
390    /// # Arguments
391    ///
392    /// * `context`: instance context
393    fn on_cluster_leader_change(&mut self, context: &PicoContext) -> CallbackResult<()> {
394        _ = context;
395        Ok(())
396    }
397
398    /// `on_healthcheck` is a callback
399    /// that should be called to determine if the service is functioning properly
400    /// On an error instance will be poisoned
401    ///
402    /// # Unimplemented
403    ///
404    /// **WARNING** This feature is not yet implemented.
405    /// The callback is never called.
406    /// TODO.
407    fn on_health_check(&self, context: &PicoContext) -> CallbackResult<()> {
408        _ = context;
409        Ok(())
410    }
411}
412
413// ---------------------------- internal implementation ----------------------------------------------
414
415/// Safe trait for sending a service trait object between ABI boundary.
416/// Define interface like [`Service`] trait but using safe types from [`abi_stable`] crate.
417#[sabi_trait]
418pub trait ServiceStable {
419    fn on_health_check(&self, context: &PicoContext) -> RResult<(), ()>;
420    fn on_start(&mut self, context: &PicoContext, configuration: RSlice<u8>) -> RResult<(), ()>;
421    fn on_stop(&mut self, context: &PicoContext) -> RResult<(), ()>;
422    fn on_leader_change(&mut self, context: &PicoContext) -> RResult<(), ()>;
423    // this macro should be moved to the last method on a major release
424    #[sabi(last_prefix_field)]
425    fn on_config_change(
426        &mut self,
427        ctx: &PicoContext,
428        new_config: RSlice<u8>,
429        old_config: RSlice<u8>,
430    ) -> RResult<(), ()>;
431    fn on_replicaset_leader_change(&mut self, context: &PicoContext) -> RResult<(), ()> {
432        _ = context;
433        ROk(())
434    }
435    fn on_cluster_leader_change(&mut self, context: &PicoContext) -> RResult<(), ()> {
436        _ = context;
437        ROk(())
438    }
439}
440
441/// Implementation of [`ServiceStable`]
442pub struct ServiceProxy<C: DeserializeOwned> {
443    service: Box<dyn Service<Config = C>>,
444}
445
446impl<C: DeserializeOwned> ServiceProxy<C> {
447    pub fn from_service(service: Box<dyn Service<Config = C>>) -> Self {
448        Self { service }
449    }
450}
451
452/// Use this function for conversion between user error and picodata internal error.
453/// This conversion forces allocations because using user-error "as-is"
454/// may lead to use-after-free errors.
455/// UAF can happen if user error points into memory allocated by dynamic lib and lives
456/// longer than dynamic lib memory (that was unmapped by system).
457fn error_into_tt_error<T>(source: impl Display) -> RResult<T, ()> {
458    let tt_error = BoxError::new(ErrorCode::PluginError, source.to_string());
459    tt_error.set_last_error();
460    RErr(())
461}
462
463macro_rules! rtry {
464    ($expr: expr) => {
465        match $expr {
466            Ok(k) => k,
467            Err(e) => return error_into_tt_error(e),
468        }
469    };
470}
471
472impl<C: DeserializeOwned> ServiceStable for ServiceProxy<C> {
473    fn on_health_check(&self, context: &PicoContext) -> RResult<(), ()> {
474        match self.service.on_health_check(context) {
475            Ok(_) => ROk(()),
476            Err(e) => error_into_tt_error(e),
477        }
478    }
479
480    fn on_start(&mut self, context: &PicoContext, configuration: RSlice<u8>) -> RResult<(), ()> {
481        let configuration: C = rtry!(rmp_serde::from_slice(configuration.as_slice()));
482        match self.service.on_start(context, configuration) {
483            Ok(_) => ROk(()),
484            Err(e) => error_into_tt_error(e),
485        }
486    }
487
488    fn on_stop(&mut self, context: &PicoContext) -> RResult<(), ()> {
489        match self.service.on_stop(context) {
490            Ok(_) => ROk(()),
491            Err(e) => error_into_tt_error(e),
492        }
493    }
494
495    fn on_leader_change(&mut self, context: &PicoContext) -> RResult<(), ()> {
496        #[allow(deprecated)]
497        match self.service.on_leader_change(context) {
498            Ok(_) => ROk(()),
499            Err(e) => error_into_tt_error(e),
500        }
501    }
502
503    fn on_replicaset_leader_change(&mut self, context: &PicoContext) -> RResult<(), ()> where {
504        match self.service.on_replicaset_leader_change(context) {
505            Ok(_) => ROk(()),
506            Err(e) => error_into_tt_error(e),
507        }
508    }
509
510    fn on_cluster_leader_change(&mut self, context: &PicoContext) -> RResult<(), ()> {
511        match self.service.on_cluster_leader_change(context) {
512            Ok(_) => ROk(()),
513            Err(e) => error_into_tt_error(e),
514        }
515    }
516
517    fn on_config_change(
518        &mut self,
519        ctx: &PicoContext,
520        new_config: RSlice<u8>,
521        old_config: RSlice<u8>,
522    ) -> RResult<(), ()> {
523        let new_config: C = rtry!(rmp_serde::from_slice(new_config.as_slice()));
524        let old_config: C = rtry!(rmp_serde::from_slice(old_config.as_slice()));
525
526        let res = self.service.on_config_change(ctx, new_config, old_config);
527        match res {
528            Ok(_) => ROk(()),
529            Err(e) => error_into_tt_error(e),
530        }
531    }
532}
533
534/// Final safe service trait object type. Can be used on both sides of ABI.
535pub type ServiceBox = ServiceStable_TO<'static, RBox<()>>;
536
537// ---------------------------- Registrar ----------------------------------------------
538
539pub type FnServiceRegistrar = extern "C" fn(registry: &mut ServiceRegistry);
540
541/// The reason for the existence of this trait is that [`abi_stable`] crate doesn't support
542/// closures.
543#[sabi_trait]
544trait Factory {
545    fn make(&self) -> ServiceBox;
546}
547
548type FactoryBox = Factory_TO<'static, RBox<()>>;
549
550/// The reason for the existence of this struct is that [`abi_stable`] crate doesn't support
551/// closures.
552struct FactoryImpl<S: Service + 'static> {
553    factory_fn: fn() -> S,
554}
555
556impl<S: Service + 'static> Factory for FactoryImpl<S> {
557    fn make(&self) -> ServiceBox {
558        let boxed = Box::new((self.factory_fn)());
559        ServiceBox::from_value(ServiceProxy::from_service(boxed), sabi_trait::TD_Opaque)
560    }
561}
562
563/// Service name and plugin version pair.
564type ServiceIdent = RTuple!(RString, RString);
565
566/// Config validator stable trait.
567///
568/// The reason for the existence of this trait is that [`abi_stable`] crate doesn't support
569/// closures.
570#[sabi_trait]
571pub trait Validator {
572    /// Validate plugin configuration.
573    ///
574    /// # Idempotency
575    ///
576    /// **WARNING** This callback may be called several times in a row.
577    /// It is the responsibility of the plugin author to make this function idempotent.
578    ///
579    fn validate(&self, config: RSlice<u8>) -> RResult<(), ()>;
580}
581
582pub type ValidatorBox = Validator_TO<'static, RBox<()>>;
583
584/// The reason for the existence of this struct is that [`abi_stable`] crate doesn't support
585/// closures.
586struct ValidatorImpl<CONFIG: DeserializeOwned + 'static> {
587    func: fn(config: CONFIG) -> CallbackResult<()>,
588}
589
590impl<C: DeserializeOwned> Validator for ValidatorImpl<C> {
591    fn validate(&self, config: RSlice<u8>) -> RResult<(), ()> {
592        let config: C = rtry!(rmp_serde::from_slice(config.as_slice()));
593        let res = (self.func)(config);
594        match res {
595            Ok(_) => ROk(()),
596            Err(e) => error_into_tt_error(e),
597        }
598    }
599}
600
601/// Registry for services. Used by picodata to create instances of services.
602#[repr(C)]
603#[derive(Default, StableAbi)]
604pub struct ServiceRegistry {
605    services: RHashMap<ServiceIdent, RVec<FactoryBox>>,
606    validators: RHashMap<ServiceIdent, ValidatorBox>,
607}
608
609impl ServiceRegistry {
610    /// Add new service to the registry.
611    ///
612    /// # Arguments
613    ///
614    /// * `name`: service name
615    /// * `plugin_version`: version of service's plugin
616    /// * `factory`: new service instance factory
617    pub fn add<S: Service + 'static>(
618        &mut self,
619        name: &str,
620        plugin_version: &str,
621        factory: fn() -> S,
622    ) {
623        let factory_inner = FactoryImpl {
624            factory_fn: factory,
625        };
626        let factory_inner =
627            FactoryBox::from_value(factory_inner, abi_stable::sabi_trait::TD_Opaque);
628
629        let ident = ServiceIdent::from((RString::from(name), RString::from(plugin_version)));
630
631        if self.validators.get(&ident).is_none() {
632            // default validator implementation,
633            // just check that configuration may be deserialized into `S::Config` type
634            let validator = ValidatorImpl {
635                func: |_: S::Config| Ok(()),
636            };
637            let validator_stable = ValidatorBox::from_value(validator, sabi_trait::TD_Opaque);
638            self.validators.insert(ident.clone(), validator_stable);
639        }
640
641        let entry = self.services.entry(ident).or_default();
642        entry.push(factory_inner);
643    }
644
645    /// Create service from service name and plugin version pair.
646    /// Return an error if there is more than one factory suitable for creating a service.
647    #[allow(clippy::result_unit_err)]
648    #[cfg(feature = "internal")]
649    pub fn make(&self, service_name: &str, version: &str) -> Result<Option<ServiceBox>, ()> {
650        let ident = ServiceIdent::from((RString::from(service_name), RString::from(version)));
651        let maybe_factories = self.services.get(&ident);
652
653        match maybe_factories {
654            None => Ok(None),
655            Some(factories) if factories.len() == 1 => {
656                Ok(factories.first().map(|factory| factory.make()))
657            }
658            Some(_) => Err(()),
659        }
660    }
661
662    /// Return true if registry contains needle service, false elsewhere.
663    /// Return an error if there is more than one factory suitable for creating a service.
664    #[allow(clippy::result_unit_err)]
665    #[cfg(feature = "internal")]
666    pub fn contains(&self, service_name: &str, version: &str) -> Result<bool, ()> {
667        let ident = ServiceIdent::from((RString::from(service_name), RString::from(version)));
668        match self.services.get(&ident) {
669            None => Ok(false),
670            Some(factories) if factories.len() == 1 => Ok(true),
671            Some(_) => Err(()),
672        }
673    }
674
675    /// Add validator for service configuration. Called before new configration is loaded.
676    /// Returning an error for validator will abort configuration change clusterwide.
677    ///
678    /// # Arguments
679    ///
680    /// * `service_name`: service name which configuration will be validated
681    /// * `plugin_version`: plugin version
682    /// * `validator`: validation function
683    pub fn add_config_validator<S: Service>(
684        &mut self,
685        service_name: &str,
686        plugin_version: &str,
687        validator: fn(S::Config) -> CallbackResult<()>,
688    ) where
689        S::Config: DeserializeOwned + 'static,
690    {
691        let ident =
692            ServiceIdent::from((RString::from(service_name), RString::from(plugin_version)));
693
694        let validator = ValidatorImpl { func: validator };
695        let validator_stable = ValidatorBox::from_value(validator, sabi_trait::TD_Opaque);
696        self.validators.insert(ident, validator_stable);
697    }
698
699    /// Remove config validator for service.
700    #[cfg(feature = "internal")]
701    pub fn remove_config_validator(
702        &mut self,
703        service_name: &str,
704        version: &str,
705    ) -> Option<ValidatorBox> {
706        let ident = ServiceIdent::from((RString::from(service_name), RString::from(version)));
707        self.validators.remove(&ident).into_option()
708    }
709
710    /// Return a registered list of (service name, plugin version) pairs.
711    #[cfg(feature = "internal")]
712    pub fn dump(&self) -> Vec<(String, String)> {
713        self.services
714            .keys()
715            .map(|key| {
716                let service = key.0.to_string();
717                let version = key.1.to_string();
718                (service, version)
719            })
720            .collect()
721    }
722}
723
724/// Migration context validator stable trait.
725///
726/// The reason for the existence of this trait is that [`abi_stable`] crate doesn't support
727/// closures.
728#[sabi_trait]
729pub trait MigrationContextValidator {
730    /// Validate the whole plugin migration context.
731    ///
732    /// # Idempotency
733    ///
734    /// **WARNING** This callback may be called several times in a row.
735    /// It is the responsibility of the plugin author to make this function idempotent.
736    fn validate(&self, context: RHashMap<RString, RString>) -> RResult<(), ()>;
737
738    /// Validate a single parameter value from the migration context.
739    ///
740    /// # Idempotency
741    ///
742    /// **WARNING** This callback may be called several times in a row.
743    /// It is the responsibility of the plugin author to make this function idempotent.
744    fn validate_parameter(&self, name: RString, value: RString) -> RResult<(), ()>;
745}
746
747pub type MigrationContextValidatorBox = MigrationContextValidator_TO<'static, RBox<()>>;
748
749/// The reason for the existence of this struct is that [`abi_stable`] crate doesn't support
750/// closures.
751#[derive(Default)]
752struct MigrationContextValidatorImpl {
753    #[allow(clippy::type_complexity)]
754    /// Whole migration context validation function. It's only argument is a
755    /// `HashMap`, that maps migration context parameter names to their values.
756    context_validator: Option<fn(context: HashMap<String, String>) -> CallbackResult<()>>,
757    /// Parameter validation function. It's first argument is the parameter name
758    /// that is currently being validated, the second argument is its value.
759    parameter_validator: Option<fn(name: String, value: String) -> CallbackResult<()>>,
760}
761
762impl MigrationContextValidator for MigrationContextValidatorImpl {
763    fn validate(&self, context: RHashMap<RString, RString>) -> RResult<(), ()> {
764        let Some(context_validator) = self.context_validator else {
765            // the context validator is not set, so don't do any validation
766            return ROk(());
767        };
768        let context = context
769            .into_iter()
770            .map(|tuple| (tuple.0.into(), tuple.1.into()))
771            .collect();
772        match (context_validator)(context) {
773            Ok(_) => ROk(()),
774            Err(e) => error_into_tt_error(e),
775        }
776    }
777
778    fn validate_parameter(&self, name: RString, value: RString) -> RResult<(), ()> {
779        let Some(parameter_validator) = self.parameter_validator else {
780            // the parameter validator is not set, so don't do any validation
781            return ROk(());
782        };
783        match (parameter_validator)(name.into(), value.into()) {
784            Ok(_) => ROk(()),
785            Err(e) => error_into_tt_error(e),
786        }
787    }
788}
789
790/// A struct that holds all entities used for migration validation (currently, only the
791/// migration context validator)
792#[repr(C)]
793#[derive(StableAbi, Default)]
794pub struct MigrationValidator {
795    context_validator: ROption<MigrationContextValidatorBox>,
796}
797
798impl MigrationValidator {
799    /// Set the whole migration context validation function. The only argument of `context_validator_fn`
800    /// is a `HashMap`, that maps migration context parameter names to their values.
801    ///
802    /// The validator function is called on `ALTER PLUGIN ... ENABLE`.
803    pub fn set_context_validator(
804        &mut self,
805        context_validator_fn: fn(context: HashMap<String, String>) -> CallbackResult<()>,
806    ) {
807        match &mut self.context_validator {
808            ROption::RNone => {
809                let validator_inner = MigrationContextValidatorImpl {
810                    context_validator: Some(context_validator_fn),
811                    parameter_validator: None,
812                };
813                let validator_inner = MigrationContextValidatorBox::from_value(
814                    validator_inner,
815                    abi_stable::sabi_trait::TD_CanDowncast,
816                );
817
818                self.context_validator = ROption::RSome(validator_inner);
819            }
820            ROption::RSome(mcv) => {
821                let mcv_impl = mcv
822                    .obj
823                    .downcast_as_mut::<MigrationContextValidatorImpl>()
824                    .expect(
825                    "downcasting should not fail, because it uses the same struct for creating the obj",
826                );
827
828                mcv_impl.context_validator = Some(context_validator_fn);
829            }
830        }
831    }
832
833    /// Set the migartion context parameter validation function. First argument of `parameter_validator_fn`
834    /// is the parameter name that is currently being validated, the second is its value.
835    ///
836    /// The validator function is called on `ALTER PLUGIN ... SET migration_context.<name>='<value>'`.
837    pub fn set_context_parameter_validator(
838        &mut self,
839        parameter_validator_fn: fn(name: String, value: String) -> CallbackResult<()>,
840    ) {
841        match &mut self.context_validator {
842            ROption::RNone => {
843                let validator_inner = MigrationContextValidatorImpl {
844                    context_validator: None,
845                    parameter_validator: Some(parameter_validator_fn),
846                };
847                let validator_inner = MigrationContextValidatorBox::from_value(
848                    validator_inner,
849                    abi_stable::sabi_trait::TD_CanDowncast,
850                );
851
852                self.context_validator = ROption::RSome(validator_inner);
853            }
854            ROption::RSome(mcv) => {
855                let mcv_impl = mcv
856                    .obj
857                    .downcast_as_mut::<MigrationContextValidatorImpl>()
858                    .expect(
859                    "downcasting should not fail, because it uses the same struct for creating the obj",
860                );
861
862                mcv_impl.parameter_validator = Some(parameter_validator_fn);
863            }
864        }
865    }
866
867    #[cfg(feature = "internal")]
868    pub fn migration_context_validator(&mut self) -> MigrationContextValidatorBox {
869        if let ROption::RSome(mcv) = self.context_validator.take() {
870            return mcv;
871        }
872
873        // context validator is not set, return a default implementation, that just returns Ok
874        let validator = MigrationContextValidatorImpl::default();
875        let validator_stable =
876            MigrationContextValidatorBox::from_value(validator, sabi_trait::TD_Opaque);
877        validator_stable
878    }
879}
880
881pub type FnMigrationValidator = extern "C" fn(validator: &mut MigrationValidator);
picodata_plugin/plugin/interface.rs

picodata_plugin/plugin/
interface.rs
