/drasi

Custom plugins (Sources, Reactions, language extensions)

This bundle covers authoring custom Drasi plugins - Sources, Reactions, and query-language extensions - against the drasi-plugin-sdk and the drasi-core engine. Pin all trait shapes shown here against the SDK version recorded in versions.md before merging code; the plugin ABI is under active rewrite (drasi-core issues #298 Plugin API, #349 Runtime plugins, #346/#362 WAL plugin, #345/#369/#392/#393/#394 Replayable Sources / Resumable Reactions / Durable Pipeline).

When to use this bundle

• Building a new Source for an upstream system that no built-in Source covers (e.g. a vendor SaaS webhook, a custom CDC stream, an in-process data feed).
• Building a new Reaction kind (a new transport, a new downstream contract, a new identity model).
• Adding a new scalar or aggregating function to the Cypher / GQL surface.
• Embedding Drasi via drasi-lib and wanting to ship Sources / Reactions inside the same binary rather than the standalone server.

If the goal is to consume an existing built-in Source or Reaction, this bundle is the wrong entry point - go to bundles/sources/guide.md or bundles/reactions/guide.md instead.

Two trait surfaces, one Source

Drasi splits "what a Source is" into two trait surfaces, and a new Source must implement both:

1. drasi_lib::sources::Source - the runtime trait the query engine talks to (ingestion, lifecycle, dispatch).
2. drasi_plugin_sdk::descriptor::SourcePluginDescriptor - the metadata / factory trait the plugin loader (or drasi-lib's static PluginRegistration) talks to to construct a Source from a JSON config.

Reactions follow the exact same split: drasi_lib::reactions::Reaction + drasi_plugin_sdk::descriptor::ReactionPluginDescriptor.

A Bootstrap provider follows the same pattern via BootstrapPluginDescriptor - used for things like state-store backends that the engine constructs before Sources / Reactions start.

Source runtime trait (drasi_lib::sources::Source)

The trait shape captured from drasi-core/components/sources/README.md on main (verify against the SDK pinned in versions.md before authoring):

• fn id(&self) -> &str - the source instance ID.
• fn type_name(&self) -> &str - provider kind (e.g. "my-source").
• fn properties(&self) -> HashMap<String, serde_json::Value> - introspection snapshot for the management API.
• fn auto_start(&self) -> bool (default true).
• fn dispatch_mode(&self) -> DispatchMode - Channel (single consumer, backpressure-aware) or Broadcast (fan-out, no backpressure).
• async fn initialize(&self, context: SourceRuntimeContext) - the single entry point. SourceRuntimeContext carries the SourceChange sender, an optional state-store handle, and an FfiLogger when loaded as a cdylib.

The supporting types live in drasi-query-ast:

• SourceChange { Insert { element }, Update { element }, Delete { reference } }
• Element { Node { metadata, properties }, Relation { metadata, in_node, out_node, properties } }
• ElementMetadata { reference: ElementReference, labels: Arc<[Arc<str>]>, effective_from: u64 }
• ElementReference::new(source_id, element_id)
• ElementPropertyMap - constructible from serde_json::Value.

A minimal SourceChange::Insert looks like:

query.process_source_change(SourceChange::Insert {
    element: Element::Node {
        metadata: ElementMetadata {
            reference: ElementReference::new("", "v1"),
            labels: Arc::new([Arc::from("Vehicle")]),
            effective_from: 0,
        },
        properties: ElementPropertyMap::from(json!({
            "plate": "AAA-1234",
            "color": "Blue"
        })),
    },
}).await;

The graph is read-only from the query side: it is mutated only by SourceChange events. There is no CREATE / MERGE / DELETE / SET from a ContinuousQuery - those clauses are inapplicable in Drasi by design.

Source plugin descriptor (SourcePluginDescriptor)

The descriptor is the factory the loader calls. Shape from drasi-plugin-sdk 0.4.2:

#[async_trait]
impl SourcePluginDescriptor for MySourceDescriptor {
    fn kind(&self) -> &str { "my-source" }
    fn config_version(&self) -> &str { "1.0.0" }
    fn config_schema_json(&self) -> String {
        let api = MySourceSchemas::openapi();
        serde_json::to_string(&api.components.as_ref().unwrap().schemas).unwrap()
    }
    fn config_schema_name(&self) -> &str { "MySourceConfigDto" }

    async fn create_source(
        &self,
        id: &str,
        config_json: &serde_json::Value,
        auto_start: bool,
    ) -> anyhow::Result<Box<dyn drasi_lib::sources::Source>> {
        let dto: MySourceConfigDto = serde_json::from_value(config_json.clone())?;
        let mapper = DtoMapper::new();
        let host = mapper.resolve_string(&dto.host)?;
        let port = mapper.resolve_typed(&dto.port)?;
        Ok(Box::new(MySource::new(id, host, port, auto_start)))
    }
}

Config DTO contract:

#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct MySourceConfigDto {
    #[schema(value_type = ConfigValueString)]
    pub host: ConfigValue<String>,
    #[schema(value_type = ConfigValueU16)]
    pub port: ConfigValue<u16>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schema(value_type = Option<ConfigValueU32>)]
    pub timeout_ms: Option<ConfigValue<u32>>,
}

ConfigValue<T> carries fields that may resolve from a literal, an environment variable, or a secret reference; DtoMapper does the resolution at create_source time. The utoipa::ToSchema derive lets drasi-host-sdk merge the plugin schema into the host's OpenAPI document served from /api/v1/openapi.json.

Reaction runtime trait (drasi_lib::reactions::Reaction)

#[async_trait]
pub trait Reaction: Send + Sync {
    fn id(&self) -> &str;
    fn type_name(&self) -> &str;
    fn properties(&self) -> HashMap<String, serde_json::Value>;
    fn query_ids(&self) -> Vec<String>;
    fn auto_start(&self) -> bool { true }
    async fn initialize(&self, context: ReactionRuntimeContext);
}

ReactionRuntimeContext carries the receive end of one ResultChange channel per subscribed query, an optional StateStore handle, and an FfiLogger when loaded as a cdylib.

The descriptor surface is symmetric to SourcePluginDescriptor, with create_reaction(...) -> anyhow::Result<Box<dyn drasi_lib::reactions::Reaction>>.

Delivery semantics a Reaction author must design for

Packaging per runtime

Supply-chain note (2026-08-20, drasi-core PR #764): the workspace removed the vulnerable h2 0.3 dependency; gRPC plugins now build on Tonic/Prost 0.14 and AWS plugins use the Hyper 1 transport. Rebuild dynamic plugins against the drasi-core release matching your host so transitive dependency versions align at load time.

A. drasi-lib (in-process, embedded)

Static linking - no FFI, no cdylib, no allocator boundary. Pass descriptors through PluginRegistration and Source / Reaction instances through the builder:

use drasi_plugin_sdk::prelude::*;
use drasi_lib::{DrasiLib, Query};

pub fn register() -> PluginRegistration {
    PluginRegistration::new()
        .with_source(Box::new(MySourceDescriptor))
        .with_bootstrapper(Box::new(MyBootstrapDescriptor))
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let source = my_source::MySource::new("sensors", config)?;
    let reaction = my_reaction::MyReaction::new("alerts", vec!["high-temp".into()]);
    let core = DrasiLib::builder()
        .with_id("my-app")
        .with_source(source)
        .with_reaction(reaction)
        .with_query(
            Query::cypher("high-temp")
                .query("MATCH (s:Sensor) WHERE s.temperature > 75 RETURN s.id, s.temperature")
                .from_source("sensors")
                .build()
        )
        .build()
        .await?;
    core.start().await?;
    tokio::signal::ctrl_c().await?;
    core.stop().await?;
    Ok(())
}

B. Drasi Server (standalone)

Build the plugin as a cdylib:

drasi_plugin_sdk::export_plugin!(
    plugin_id = "my-source",
    core_version = env!("CARGO_PKG_VERSION"),
    lib_version = env!("CARGO_PKG_VERSION"),
    plugin_version = env!("CARGO_PKG_VERSION"),
    source_descriptors = [MySourceDescriptor],
    reaction_descriptors = [],
    bootstrap_descriptors = [MyBootstrapDescriptor],
);

The macro emits the FFI entry points (drasi_plugin_metadata, drasi_plugin_init), a dedicated tokio runtime inside the cdylib, and FfiLogger / lifecycle callback wiring. The host's tokio is not shared across the FFI boundary - the plugin owns its own.

The resulting libdrasi_source_my.{so,dylib,dll} drops into the Drasi Server plugin directory. The server's PluginLoader (in drasi-host-sdk) dlopens it, validates SDK major.minor and target triple, then calls init(). Target-triple mismatch silently fails - match the host's exactly (Linux gnu / Linux musl / Windows gnu - see drasi-core/Cross.toml for the supported matrix).

C. Drasi for Kubernetes (drasi-platform)

Built-in providers live under drasi-platform/sources/ and drasi-platform/reactions/. After PR #298 + #349 landed, the K8s runtime uses the same plugin SDK as Drasi Server. The path for a new provider is: implement the Source / Reaction + descriptor pair, build as a cdylib, register the provider type in drasi-platform. CRD schemas (Source / Query / Reaction) and identity wiring (Microsoft Entra Workload ID for Azure-aware providers) follow drasi.io/drasi-kubernetes/how-to-guides/.

Lifecycle (cdylib loading)

1. Host starts with the dynamic-plugins feature.
2. PluginLoader scans the plugin directory.
3. For each .so / .dylib / .dll: a. dlopen() the shared library. b. Resolve drasi_plugin_metadata() → PluginMetadata. c. Validate SDK version (major.minor must match host). d. Validate target triple (must match host). e. Resolve drasi_plugin_init() → FfiPluginRegistration. f. Call init → plugin spins up its own tokio runtime + installs FfiLogger. g. Host wraps the returned vtables as SourceProxy / ReactionProxy / SourcePluginProxy implementing the drasi_lib runtime traits.
4. User POSTs config to the management API → host calls create_source(id, config_json, auto_start) on the descriptor proxy, gets back Box<dyn drasi_lib::sources::Source>, registers it with the running DrasiLib.
5. DrasiLib::start() calls Source::initialize(ctx); the source begins ingesting and pushing SourceChange events into the engine.
6. DrasiLib::stop() triggers shutdown; the source drains in-flight changes and returns.

Plugin discovery and naming rules

• The scan keys on the shared drasi_ filename token, not the plugin type. As of drasi-core PR #662 (merged 2026-08), the host's directory scan matches plugins by their common drasi_ prefix and is type-agnostic; it no longer infers Source vs Reaction from per-type filename patterns. A library without the drasi_ token is silently skipped at scan time. Keep the built-in naming convention (libdrasi_source_<name>.so, libdrasi_reaction_<name>.dll) so your cdylib matches on every platform's lib-prefix quirks.
• Nightly plugin crate builds exist. drasi-core CI publishes plugin crates on a nightly schedule (publishing workflow cleaned up in PR #726). Useful for testing custom plugins against unreleased engine changes; pin exact nightly versions in test configs only, never float them in production configs.

Cross-cdylib Box ownership pitfall (drasi-core issue #378)

Symptom. A Box<dyn Trait> returned from a cdylib plugin and dropped in the host (or vice versa) can use a different allocator and vtable layout than the receiver expects, leading to undefined behaviour - heap corruption, double-free, or silent memory leak.

Scope. This affects every dynamic plugin entry point that returns Box<dyn drasi_lib::sources::Source> or Box<dyn drasi_lib::reactions::Reaction> from a create_* method - i.e. the entire dynamic plugin API surface on main as of 2026-05-15. Issue #378 is open.

Mitigations to apply today.

• Build plugins from the same Cargo workspace and the same rustc toolchain as drasi-server and drasi-host-sdk. Vendor in via [patch.crates-io] or path dependencies; do not float versions against crates.io alone.
• Pin drasi-plugin-sdk and drasi-host-sdk to identical exact versions in the workspace Cargo.toml. Loader validates SDK major.minor; allocator safety requires the exact patch + toolchain match.
• Match the target triple the host validates at load time. If the host is Linux musl, the plugin MUST be Linux musl too.
• Pin Rust toolchain via rust-toolchain.toml to the version drasi-core uses (currently 1.90 after PR #409).
• Do not use a system allocator override (#[global_allocator]) in the plugin crate.

Long-term direction (hypothesis, not in the codebase yet). A stable C-ABI factory function with explicit extern "C" fn drop_* callbacks rather than naive Box<dyn Trait> across the boundary; re-verify when #378 closes.

Async runtime contract

• A dynamic plugin owns its own tokio runtime. The export_plugin! macro generates the runtime inside the cdylib. The host's tokio is not shared across FFI.
• An embedded drasi-lib plugin uses the host's tokio directly - there is no FFI.
• Engine channels (SourceChange, ResultChange) are bounded by dispatch_buffer_capacity (default 1,000) and priority_queue_capacity (default 10,000). Saturate either and the source/reaction blocks (Channel mode) or drops (Broadcast).
• Logging from a dynamic plugin flows through FfiLogger and the host's host_log_callback. Do not write directly to stdout from a cdylib plugin - log lines bypass the host's structured logger.

Identity / Workload ID wiring

The runtime application-identity provider landed in drasi-core PR #423 (2026-05-12). A new Reaction that targets an Azure service (Event Grid, SignalR, Service Bus, Cosmos DB) should acquire credentials through that provider rather than reading environment variables directly - that is the surface that bridges to Microsoft Entra Workload ID for the K8s runtime. Confirm the trait shape on the SDK version pinned in versions.md before merging code.

Built-in plugin templates to read first

Query-language extension points

Adding new behaviour to the query language has a strict ordering of "easiest" to "hardest". Prefer the cheapest extension that solves the problem.

Add a new scalar function

1. Define a function module under core/src/evaluation/functions/scalar/ implementing the scalar trait (sync) or LazyScalarFunction (deferred).
2. Register it in functions-cypher/src/ so with_cypher_function_set() picks it up.
3. Register it in functions-gql/src/ separately. The two registries are independent - a function registered in one is invisible from the other (tracked in drasi-core issue #217).
4. No AST change needed. FunctionExpression { name, args, position_in_query } already represents any named call; dispatch is by name.

Add a new aggregating function

The well-grounded path. Implement the AggregatingFunction trait:

#[async_trait]
pub trait AggregatingFunction {
    fn initialize_accumulator(
        &self,
        context: &ExpressionEvaluationContext,
        expression: &ast::FunctionExpression,
        grouping_keys: &Vec<VariableValue>,
        index: Arc<dyn ResultIndex>,
    ) -> Accumulator;

    async fn apply(
        &self,
        context: &ExpressionEvaluationContext,
        args: Vec<VariableValue>,
        accumulator: &mut Accumulator,
    ) -> Result<VariableValue, FunctionError>;

    async fn revert(
        &self,
        context: &ExpressionEvaluationContext,
        args: Vec<VariableValue>,
        accumulator: &mut Accumulator,
    ) -> Result<VariableValue, FunctionError>;

    async fn snapshot(
        &self,
        context: &ExpressionEvaluationContext,
        args: Vec<VariableValue>,
        accumulator: &Accumulator,
    ) -> Result<VariableValue, FunctionError>;

    fn accumulator_is_lazy(&self) -> bool;
}

• apply runs when a row newly matches or is updated-to-match.
• revert runs when a row newly stops matching.
• snapshot returns the current value without mutating.
• This is what makes Drasi aggregations correct under continuous change - values are individually applied and reverted, not recomputed from scratch.

Procedure:

1. Create core/src/evaluation/functions/aggregation/my_fn.rs.
2. Implement the trait above.
3. Choose accumulator storage:
   • Simple stateful (sum / count / list) → Accumulator::Value( ValueAccumulator::...). Add a ValueAccumulator variant in aggregation/mod.rs if no existing variant (Sum, Avg, Count, LinearGradient, Value, Map, TimeMarker, Signature) fits.
   • Order-dependent (top-K, percentile) → Accumulator::LazySortedSet(LazySortedSet).
4. Register in functions-cypher and functions-gql separately.
5. Add a use-case test under shared-tests/src/use_cases/<my_fn>/mod.rs mirroring sensor_heartbeat/. The test is re-run automatically across in-memory, RocksDB, Garnet, and redb-WAL backends via the QueryTestConfig trait.

Add a new operator

Operators are AST nodes - UnaryExpression and BinaryExpression in query-ast/src/ast.rs. Adding a new operator requires four changes:

1. Extend the UnaryOperator / BinaryOperator enum in query-ast.
2. Extend the query-cypher grammar to emit the new operator.
3. Extend the query-gql grammar to emit the same AST node.
4. Extend the ExpressionEvaluator dispatch in core/src/evaluation/expressions/mod.rs to handle the new variant.

This is materially harder than adding a function. In almost all cases the right answer is to add a function, not an operator.

Add a new clause

The heaviest extension. Adding a new clause (e.g. a hypothetical WATCH (n) FOR 5m syntax) requires:

1. New QueryPart variant in query-ast.
2. Parser support in both query-cypher and query-gql.
3. Planner / evaluator extension under core/src/evaluation/parts/.
4. Storage interaction (likely a new ResultIndex subtrait or a new FutureQueue use).
5. Backend implementations across in-memory, RocksDB, and Garnet.

The Drasi convention is to express temporal / windowing logic as drasi.* function calls - drasi.trueFor(predicate, duration), drasi.previousValue(prop), drasi.getVersionByTimestamp(elem, ts). These are dispatched through FunctionRegistry and a FutureQueue, not a new clause. New temporal logic should follow this pattern.

Decision matrix: fork drasi-core or use a downstream extension point?

Re-verify the scalar / aggregating function row before assuming a fork is required - runtime-plugin work in drasi-core PR #349 may have added a with_function(...) extension point not present at the time of writing.

Replayable Sources / Resumable Reactions status

Tracked in drasi-core issues #345, #369, #392, #393, #394; WAL plugin contract + redb implementation merged in #346 / #362 on or about 2026-05-12.

Goal: Sources persist a Write-Ahead Log so that on restart, the engine can replay from a known offset rather than re-bootstrap from the upstream system.

Implications for new plugin authors today.

• Until #392-#394 land, design Sources to be idempotent on replay - the engine may re-deliver SourceChange events after a crash; do not assume exactly-once delivery into the engine.
• The Source trait signature will likely change again to thread WAL offsets through initialize(ctx). Pin a drasi-plugin-sdk minor version in versions.md; re-verify trait shapes on each upgrade.
• Reaction authors should keep consumers idempotent on duplicate delivery - Resumable Reactions are part of the same work stream.

Test patterns

Non-negotiable rules for a custom plugin

1. Pin drasi-plugin-sdk and drasi-host-sdk to exact identical versions in the workspace Cargo.toml. SDK major.minor mismatch is validated by the loader and rejected; allocator safety requires exact patch + toolchain match (issue #378).
2. Match the target triple of the host - Linux gnu, Linux musl, or Windows gnu. Mismatch silently fails to load.
3. Design for at-least-once + idempotent until Replayable Sources / Resumable Reactions ship.
4. Do not return Box<dyn Trait> allocated under a different allocator than the host's. Build inside the drasi-core workspace; do not set #[global_allocator] in the plugin crate.
5. Pin the Rust toolchain via rust-toolchain.toml to the version drasi-core uses (currently 1.90).
6. Capture the SDK version, drasi-core commit SHA, and target triple in versions.md. The cross-runtime CI gate must fail if these drift from the host the plugin is deployed against.
7. Add a shared-tests/src/use_cases/<name>/mod.rs test for any language extension (function / operator / clause). Unit tests are not sufficient - the test must run against multiple storage backends.
8. Log through FfiLogger, not stdout, in any cdylib plugin.
9. Choose DispatchMode deliberately - Channel for correctness-critical Sources / Reactions, Broadcast only for fan-out where slow consumers may drop.
10. Do not assume a stable plugin ABI on main - three of the four major surfaces (Plugin API #298, Runtime plugins #349, WAL plugin #346/#362, Replayable Sources / Resumable Reactions #345/#369/#392/#393/#394) are mid-rewrite. Re-verify against the SDK version in versions.md before every merge.

Acceptance evidence for a new plugin

Capture each of these on the PR that lands a new Source, Reaction, or language extension. Cross-reference from the templates/drasi-acceptance-evidence.md record:

• Built artefact and target triple (cdylib for Drasi Server / K8s, static crate for drasi-lib); SHA-256 of the artefact.
• SDK version pin (drasi-plugin-sdk = "X.Y.Z") and host version pin captured in versions.md.
• Rust toolchain pin (rust-toolchain.toml matches drasi-core).
• cargo test output for the plugin crate (unit layer).
• shared-tests output for any language extension, captured across every storage backend the upstream test suite runs (in-memory, RocksDB, Garnet, redb-WAL).
• lib-integration-tests output if the plugin touches cross-cutting behaviour.
• Loader log lines proving SDK validation passed (Validate SDK version: ok, Validate target triple: ok) - capture from the host log on a clean start.
• Replay safety statement: explicit one-line claim that the Source is idempotent on SourceChange re-delivery, and the Reaction is idempotent on ResultChange re-delivery. Cross-reference the property in versions.md.
• Cross-cdylib ownership statement (Reactions / Sources only): confirmation the plugin was built from the drasi-core workspace with no #[global_allocator] override.
• For Azure-aware Reactions: confirmation that credentials are acquired through the application-identity provider (drasi-core PR #423), not from environment variables.
• Updated bundles/custom-plugins/guide.md cross-references in the PR if the new plugin reveals a trait-shape change worth recording here.