extension-system-architecture.md

Extension System Architecture

Overview

This document describes the proposed Phase 1 architecture for the extension system in the OTAP dataflow engine, building on the extension system proposal which establishes the vision, goals, and phased rollout plan.

A working proof of concept is available on the PoC branch.

How this document relates to the proposal

The proposal defines what the extension system should do and why. This document describes how each requirement is addressed in the Phase 1 implementation:

Proposal Requirement	Phase 1 Approach
Capability-based access	#[capability] proc macro generates typed traits; consumers resolve via require_local() / require_shared()
Multiple implementations of same capability	CapabilityRegistry keyed by (extension_name, TypeId) -- different extensions can provide the same capability
Multiple configured instances	extensions: section in YAML, each with a unique name; nodes bind by name in capabilities:
Existing config model integration	Extensions are siblings to nodes in the pipeline config hierarchy
Preserve performance model (thread-per-core)	Local extensions use Rc (no locks); shared extensions use Clone + Send with Arc-wrapped state
Background tasks	Active extensions get their own event loop via Extension::start()
Explicit capability binding	Nodes declare capabilities: { name: extension_instance } -- no implicit discovery
No hot-path registry lookup	Capabilities resolved once at factory time; nodes hold typed handles for their lifetime
Future hierarchical scopes	CapabilityRegistry and resolve_bindings() are scope-agnostic by design

Beyond the proposal's requirements, the design rests on four additional principles:

• Pipeline-scoped extensions are per-core. Both local and shared pipeline-scoped extensions are instantiated per pipeline instance (i.e., per core). The local/shared distinction at pipeline scope is about type constraints (!Send vs Send + Clone), not about cross-core sharing. This follows a consistent principle: an extension's sharing boundary is determined by the scope it is declared in, not by its execution model. Pipeline is the only scope in Phase 1; broader scopes (group, engine) and narrower scopes (node-set, node) are future work and will each define their own sharing boundary. The execution model (local vs shared) determines only the type constraints imposed on the implementation.
• Active/Passive lifecycle distinction. Not every extension needs a background task. Extensions that only provide capabilities are marked Passive -- no task is spawned, no control channel is allocated, no shutdown messages are sent, with zero runtime overhead. Extensions that drive their own event loop are marked Active and receive a task, a control channel, and shutdown orchestration.
• Instance policy (Cloned vs Constructed). A capability consumer receives either a clone of a stored prototype (.cloned()) or a newly-constructed instance from a user-supplied closure (.constructed()). The policy is chosen by the extension author at build time and is invisible to consumers -- require_shared() / require_local() return the same trait-object type regardless. .constructed() is Passive-only; the combination Active + Constructed is unrepresentable in the typestate builder (an Active extension has a single engine-driven event loop that doesn't compose with per-consumer construction).
• Shared-to-local transparent fallback. A shared-only extension can serve local consumers via a wrap_shared_as_local adapter (generated by the #[capability] proc macro). Most extensions only need a shared implementation; local consumers are served automatically. Extension authors who want the lock-free performance of a local-only implementation (e.g., Rc<RefCell> instead of Arc<RwLock>) can opt in by providing a dedicated local variant.

Ownership and cloning model

Local capabilities return Rc<dyn local::Trait> -- all local consumers share the same instance via reference counting. No cloning, no locks.

Shared capabilities return Box<dyn shared::Trait> -- each consumer gets an independent clone. For shared extensions to share mutable state across clones (e.g., token senders, connection pools), fields should be wrapped in Arc, similar to how tokio, axum, and reqwest handle shared state:

#[derive(Clone)]
struct MyExtension {
    // Shared mutable state -- Arc ensures clones
    // see the same data
    token_sender: Arc<watch::Sender<Option<BearerToken>>>,
    credential: Arc<dyn TokenCredential>,
    // Plain data -- cloned independently per consumer
    scope: String,
}

What Are Extensions?

Extensions are standalone pipeline components that provide shared, cross-cutting capabilities -- such as authentication, storage etc. -- to data-path nodes (receivers, processors, exporters). They are configured as siblings to nodes, not as nodes themselves, and they never touch pipeline data directly.

Architecture Overview

+----------------------------------------------------------+
|                     Pipeline Engine                      |
|                                                          |
|  +-------------------+  +-------------------+            |
|  | Extension A       |  | Extension B       |  ...       |
|  | Active(auth)      |  | Passive(kv store) |            |
|  | local + shared    |  | shared only       |            |
|  | lifecycle         |  | no task spawned   |            |
|  +---------+---------+  +---------+---------+            |
|            | #[capability] proc macro                    |
|            | + extension_capabilities!() macro           |
|            v                                             |
|  +----------------------------+                          |
|  |    CapabilityRegistry      |  (built once per         |
|  |  local_handles HashMap     |   pipeline)              |
|  |  shared_handles HashMap    |                          |
|  +----+-----------------+-----+                          |
|       | resolve_bindings|                                |
|       v                 v                                |
|  +-----------+  +-----------+                            |
|  | Receiver  |  | Exporter  |                            |
|  | require   |  | require   |                            |
|  | _local()  |  | _shared() |                            |
|  | -> Rc<T>  |  | -> Box<T> |                            |
|  +-----------+  +-----------+                            |
|                                                          |
|  Local consumers get Rc<dyn local::Trait>                |
|  Shared consumers get Box<dyn shared::Trait> (Send)      |
+----------------------------------------------------------+

Key Design Decisions

1. Extensions start first, shut down last. Active extensions are spawned before data-path nodes. At shutdown, extensions terminate only after all data-path nodes have drained. Passive extensions (no lifecycle) skip spawning entirely.

2. PData-free. Extensions are completely decoupled from the pipeline data type. They use ExtensionControlMsg through a dedicated control channel.

3. Active vs Passive lifecycle. Extensions choose their lifecycle at build time via a typestate builder stage: .active() or .passive(). Active extensions get a task and control channel. Passive extensions only provide capabilities -- no task is spawned, no control channel is allocated, no messages are sent. The .active() stage only accepts types implementing Extension; the .passive() stage does not require that trait.

4. Background lifecycle (zero capabilities). A third lifecycle, .background(), sits alongside .active() / .passive() for engine-driven services that expose no capability -- periodic reporters, schedulers, health monitors, global rate-limit coordinators. The builder shape is .background().shared(impl_) or .background().local(Rc::new(impl_)) followed by .build(); exactly one registration is required and a second is unrepresentable in the typestate. The choice of .shared(...) vs .local(...) only governs how the engine hosts the instance (Send + Clone vs !Send, per-pipeline). Background extensions never appear as the right-hand side of a capability binding; their factory's capabilities field is Option<_>::None, and that None is the engine's runtime signal "this is a Background extension" -- capability registration is skipped entirely. For lifecycle dispatch (event loop, control channel, shutdown sequencing) Background is handled exactly like Active. The shape constraints are compile-time enforced by the typestate builder:

   • Active and Passive must register >=1 capability.
   • Background must register 0 capabilities (no extension_capabilities! invocation).
   • Background must register exactly one of .shared(...) / .local(...) -- never both.

5. Instance policy (Passive only). After .passive(), the extension author picks an instance policy. The choice is provider-side only -- capability consumers call require_shared() / require_local() and cannot observe which policy was used.

   • .cloned() -- each consumer receives a clone of the value handed to the builder. The semantics differ by execution model:
     • For .shared(value: E) (E: Clone + Send), each consumer gets value.clone() -- an independent copy of the underlying object.
     • For .local(rc: Rc<E>), each consumer gets Rc::clone(&rc) -- a new handle to the same underlying object. Local consumers in the same pipeline instance therefore share one extension instance.
   • .constructed() -- each consumer receives a newly-constructed instance from a user-supplied Fn() -> E + Clone closure.

   Active + Constructed is unrepresentable: ActiveStage exposes no .constructed() method, so the invalid combination is a compile-time error.

6. Local/Shared split. Each lifecycle/policy stage lets the user register at most one .shared(...) variant and one .local(...) variant. A single extension can provide one or both:

   • Shared-only (with local fallback): .active().shared(ext).build() -- the shared type serves both local and shared consumers via wrap_shared_as_local fallback. This is the most common pattern.
   • Local-only: .active().local(Rc::new(ext)).build() -- only local consumers can use this extension. Shared consumers (require_shared()) get a config error. Use when the extension is inherently !Send.
   • Dual-type: .active().local(Rc::new(l)).shared(s).build() -- separate types with independent lifecycles.
   • Passive cloned: .passive().cloned().shared(ext).build() -- no lifecycle; consumers clone a stored prototype.
   • Passive constructed: .passive().constructed().shared(|| MyExt::new(cfg.clone())).build() -- no lifecycle; each consumer invokes the stored constructor closure.

7. Type-safe capability resolution. Consumers call capabilities.require_local::<BearerTokenProvider>() (returns Rc<dyn local::BearerTokenProvider>) or capabilities.require_shared::<KeyValueStore>() (returns Box<dyn shared::KeyValueStore>, which is Send). The zero-sized registration struct carries associated types (Local and Shared) that map to the correct trait object variants. Sealing via ExtensionCapability ensures only engine-defined capabilities are accepted at compile time. Each of the four accessors (require_* / optional_*) is intended to be called exactly once per capability per node at node construction; a second call returns Error::CapabilityAlreadyConsumed. Local fallback from shared extensions is materialized lazily at that call: on require_local(), the registered shared factory runs and its result is routed through the capability's wrap_shared_as_local adapter to produce Rc<dyn C::Local>.

8. #[capability] proc macro. Each capability is defined via a single #[capability] attribute on a trait definition. The macro generates: local:: and shared:: trait variants, a wrap_shared_as_local adapter fn for transparent fallback, sealed trait impls, a zero-sized registration struct, a KNOWN_CAPABILITIES link-time entry (via distributed_slice), and typed shared_entry::<E> / local_entry::<E> caster functions that bridge an extension's SharedInstanceFactory / LocalInstanceFactory into a registry entry. Consumers use trait objects directly. Shared data types (e.g., BearerToken, Secret) are hand-written alongside the macro invocation.

Module Layout

engine/src/
  lib.rs                    -> ExtensionFactory, engine build logic

  extension/
    mod.rs                  -> module root, ExtensionBundle,
                              ExtensionLifecycle, ExtensionWrapper
    builder.rs              -> Typestate builder: ActiveStage,
                              PassiveStage, PassiveClonedStage,
                              PassiveConstructedStage
    wrapper.rs              -> ExtensionWrapper variants,
                              ControlChannel, EffectHandler
    tests.rs                -> extension-level tests

  capability/
    mod.rs                  -> ExtensionCapability sealed trait,
                              KnownCapability, extension_capabilities!
                              declarative macro, KNOWN_CAPABILITIES
    factory.rs              -> SharedInstanceFactory /
                              LocalInstanceFactory (cloneable
                              type-erased produce closures encoding
                              instance policy)
    tests.rs                -> extension_capabilities! macro tests
    registry/
      mod.rs                -> public re-exports
      entry.rs              -> SharedCapabilityEntry, LocalCapabilityEntry,
                              ResolvedSharedEntry, ResolvedLocalEntry,
                              cloneable produce closures
      storage.rs            -> CapabilityRegistry HashMap store
      capabilities.rs       -> Per-node Capabilities handle
                              (require_*/optional_*)
      resolve.rs            -> resolve_bindings: validate + produce
                              per-node Capabilities
      tracker.rs            -> ConsumedTracker for unused-extension
                              accounting
      tests.rs              -> end-to-end registry tests

  local/
    extension.rs            -> Extension trait (!Send, Rc<Self>)
    capability.rs           -> re-exports (populated per-capability)
    exporter.rs, receiver.rs, processor.rs  (unchanged)

  shared/
    extension.rs            -> Extension trait (Send, Box<Self>)
    capability.rs           -> re-exports (populated per-capability)
    exporter.rs, receiver.rs, processor.rs  (unchanged)

The module dependency is one-way: extension/ depends on capability/, not the reverse. The instance factories live in capability/factory.rs because they're consumed by the capability registry's registration fn pointers; extension/builder.rs and extension/wrapper.rs import them from there.

Concrete capabilities (e.g. bearer_token_provider, key_value_store) land alongside their first consumer in subsequent PRs; the capability/ tree above holds only the registry infrastructure and the sealed ExtensionCapability trait.