Crate photon

Expand description

Pub/sub event pipeline (public crate).

Typed topics, durable subscriptions with checkpoints, and the same API in single-process and multi-node deployments. Enable the runtime feature for the full stack (Photon, backends, executor). Photon uses pluggable storage adapters (mem, sqlite, nats, fluvio, kafka) behind StoragePort and Quark inventory for topic/handler discovery. Payloads are encrypted before append; storage stays opaque. Canonical business data remains in your datastore — the transport log is for event delivery, not system of record.

§Features

Typed topics and handlers — topic / subscribe register via Quark inventory
Same API everywhere — publish_on / start_executor whether you run one process or many
Pluggable storage — mem, sqlite, or broker adapters (nats, kafka, fluvio)
Durable subscriptions — checkpointed replay after restart on durable adapters
Host-owned crypto — PHOTON_TRANSPORT_KEY seals envelopes, keeping ciphertext at rest

Typed pub/sub over a persistent event log — pick embedded or brokered topology.

§Getting started

You always publish and subscribe the same way (OrderCreated { … }.publish_on(&photon), #[subscribe], Photon::start_executor). What changes is whether events stay in one process or cross a broker to another binary.

§Choose a topology

Embedded — one process owns publish and handlers. Start here (mem or sqlite).
Brokered — publisher binary(ies) and worker binary(ies) share a broker. Photon does not ship a separate server binary; each process embeds Photon against the same adapter.

Topology	Adapter	Features	When to use
Embedded	`InProcStoragePort` (default)	`runtime`,`mem`	Local / tests
Embedded durable	`SqliteStoragePort`	`runtime`,`sqlite`	Single host, restart-safe
Brokered	NATS / Kafka / Fluvio	`runtime` + `nats`/`kafka`/`fluvio`	Multi-process / fleet

Adapter builders and env vars: config.

After you pick a topology, continue with declare topics.

§Embedded (one binary)

This process publishes and runs handlers. There is no second binary and no external broker.

Your app ──publish / #[subscribe]──► Photon ──StoragePort──► mem | sqlite

Prerequisites: Cargo features runtime + mem (or sqlite), and PHOTON_TRANSPORT_KEY (base64 of 32 bytes). See config.

In-memory (default — PhotonBuilder installs InProcStoragePort when you omit storage_port):

use std::sync::Arc;

use photon::{JsonIdentityFactory, Photon};

let photon = Photon::builder().auto_registry().build()?;
photon.start_executor(Arc::new(JsonIdentityFactory))?;
// Prefer EventType { … }.publish_on(&photon).await

SQLite — durable single-process (write-through + in-memory live fanout):

use std::sync::Arc;

use photon::{JsonIdentityFactory, Photon, SqliteStoragePort};

let port = Arc::new(SqliteStoragePort::open("/var/lib/photon/events.db").await?);
let photon = Photon::builder()
    .storage_port(port)
    .auto_registry()
    .build()?;
photon.start_executor(Arc::new(JsonIdentityFactory))?;

Runnable: cargo run -p uf-photon --example embedded_mem --features runtime,mem. Durable: cargo run -p uf-photon --example embedded_sqlite --features runtime,sqlite. Then jump to declare topics.

§Brokered (publisher + worker binaries)

Use this when multiple processes (or hosts) must share the same topic log. mem and sqlite cannot fan out across processes — wire a broker adapter instead.

Publisher binary  ──publish_on──► Photon ──StoragePort──► broker (NATS / Kafka / Fluvio)
Worker binary     ──start_executor / #[subscribe]──► Photon ──same broker──►

§What you create

Piece	Purpose
Shared topics	Same `#[topic]` types (shared crate or both binaries compile them)
Publisher binary	`[[bin]]` that publishes; typically no `start_executor`
Worker binary	`[[bin]]` with `#[subscribe]` handlers; must call `start_executor`
Broker	NATS / Kafka / Fluvio cluster your ops team runs
Shared env	Same `PHOTON_TRANSPORT_KEY` + broker URL (e.g. `PHOTON_NATS_URL`) on every process
Broker TLS	Default `BrokerTransportSecurity::RequireTls`; plaintext needs explicit opt-in

§Shared setup (both binaries)

Enable runtime plus one broker feature (nats is the usual first choice).
Build the same storage port against the same cluster — pick a builder below.
Call .auto_registry() when using macros.
Keep the Photon handle for publish_on / subscribe_on.

Adapter	Feature	Builder (publisher + worker examples)
NATS	`nats`	`NatsStoragePortBuilder`
Kafka	`kafka`	`KafkaStoragePortBuilder`
Fluvio	`fluvio`	`FluvioStoragePortBuilder`

Concern	Production setting
Transport key	`PHOTON_TRANSPORT_KEY` from a secret manager (never `PHOTON_ALLOW_DEV_TRANSPORT_KEY`)
Broker TLS	`.require_tls()` / TLS URL (`tls://…`); never set `PHOTON_ALLOW_INSECURE_BROKER`
NATS auth	`.credentials_file("/run/secrets/nats.creds")` or `PHOTON_NATS_CREDS` (not URL userinfo)
Kafka retention	Pre-create topics with `retention.ms` or broker defaults (`rskafka` cannot set configs)
Fluvio retention	Applied at topic create from `PHOTON_FLUVIO_RETENTION`

Env index: config. NATS sketches follow; swap the builder for Kafka/Fluvio.

§Publisher binary

Wire the broker port, declare topics, publish. Skip Photon::start_executor unless this process also handles events.

use std::sync::Arc;

use photon::{Photon, NatsStoragePort, ReplayCursor};

let port = Arc::new(
    NatsStoragePort::builder()
        .url("tls://nats.example:4222")
        .credentials_file("/run/secrets/nats.creds")
        .require_tls()
        .replay_cursor(ReplayCursor::StreamSeq)
        .sync_ack(true)
        .build()
        .await?,
);
let photon = Photon::builder()
    .storage_port(port)
    .auto_registry()
    .build()?;
// OrderCreated { … }.publish_on(&photon).await?;

§Worker binary

Same storage port wiring as the publisher, plus #[subscribe] handlers and Photon::start_executor:

use std::sync::Arc;

use photon::{JsonIdentityFactory, Photon, NatsStoragePort, ReplayCursor};

let port = Arc::new(
    NatsStoragePort::builder()
        .from_env_defaults()
        .replay_cursor(ReplayCursor::StreamSeq)
        .sync_ack(true)
        .build()
        .await?,
);
let photon = Photon::builder()
    .storage_port(port)
    .auto_registry()
    .build()?;
photon.start_executor(Arc::new(JsonIdentityFactory))?;

§Run both

Start the broker.
Start worker(s) first so subscriptions are ready.
Start one or more publishers.

Runnable: cargo run -p uf-photon --example nats_worker --features runtime,nats then nats_publisher (same features). Same pair contract with other adapters: kafka_worker + kafka_publisher (runtime,kafka), fluvio_worker + fluvio_publisher (runtime,fluvio), and the production-TLS variant nats_secure_worker + nats_secure_publisher (runtime,nats). Multi-terminal runbook: repository photon/README.md § How to run examples.

Then continue with declare topics.

§3. Declare topics and handlers

topic — typed event struct + inventory registration; generates publish_on / subscribe_on
subscribe — inventory-registered handler; requires Photon::start_executor on workers
prelude — common imports (Event, SubscribeOpts, Photon, macros)

Attribute tables: config. Runnable: keyed_topic, consumer_group, subscribe_v2.

After topic, the macro generates typed methods on your event struct. Prefer those over raw Photon::publish / Photon::subscribe.

Publish — publish_on with an explicit Photon handle:

OrderCreated {
    order_id: "ord-1".into(),
    amount_cents: 9900,
}
.publish_on(&photon)
.await?;

Typed stream — subscribe_on (no #[subscribe] / executor required):

use futures::StreamExt;
use photon::SubscribeOpts;

let opts = SubscribeOpts::default_ephemeral();
let mut stream = OrderCreated::subscribe_on(&photon, opts).await?;
if let Some(Ok(envelope)) = stream.next().await {
    let _payload = envelope.payload; // OrderCreated
}

Inventory handlers — #[subscribe] + Photon::start_executor (Embedded hosts and Brokered workers).

Optional sugar after configure: .publish() / .subscribe() without a handle. Raw topic-name API (advanced): Photon::subscribe.

Runnable: embedded_mem (publish_on + #[subscribe]), keyed_topic (subscribe_on), manual_subscribe (raw Photon::subscribe).

§5. Run the executor and reclaim

Photon::start_executor — dispatch inventory-registered #[subscribe] handlers (Embedded hosts and Brokered workers)
Photon::reclaim_transport — retention sweep past the safe watermark
Optional ops telemetry: OpsLog via PhotonBuilder::ops_log (example: telemetry_ops_log)
Restart-safe checkpoints: durable = "…" resumes from the last committed seq after a crash (example: durable_consumer_recovery, runtime,sqlite)

§Notes

Transport key: boot fails closed without a valid PHOTON_TRANSPORT_KEY (see config). Actor/payload JSON is sealed before storage and broker write; handlers receive decrypted events.
Host responsibilities: Photon is a trusted-host library — authenticate and authorize at the process edge before publish/subscribe/admin/WS. See repository SECURITY.md (privileged handle, no in-core topic ACL, production identity factory, broker TLS via BrokerTransportSecurity / PHOTON_ALLOW_INSECURE_BROKER, never PHOTON_ALLOW_DEV_TRANSPORT_KEY in production).
Durable multi-process: use a broker adapter; mem does not cross process boundaries.
Lab topologies: testkit / bench PHOTON_TOPOLOGY values are harness labels for lab matrices. Product choices remain Embedded / Brokered above.
Custom adapters: implement StoragePort and pass it to PhotonBuilder::storage_port. Advanced delivery traits live behind that port.

§Architecture

Application  →  Photon (macros + Photon runtime)  →  storage port  →  delivery backend

  Typed API (#[topic], publish_on / publish, #[subscribe])
             │
             v
  Photon runtime
             │
             v
        StoragePort  ──►  mem (InProcStoragePort)
                     ──►  sqlite (SqliteStoragePort)
                     ──►  nats / fluvio / kafka (broker crates)
                     ──►  custom implementation

Full option reference: config. Macro expansion: repository docs/macro-expansion.md.

Re-exports§

pub use quark::inventory;

Modules§

backend: Backend implementations and assembly context for Photon.
backend_instrumentation: Backend instrumentation (L0 publish telemetry + ops log helpers).
checkpoint: Checkpoint coalescing for durable subscriptions.
config: Photon configuration reference
consumer_group: Consumer group coordination: shard assignment and leases.
delivery: Handler delivery: backpressure and dead-letter queue.
delivery_mode: Topic delivery mode and virtual shard configuration.
descriptor: Topic descriptor for auto-registration.
error: Error types for Photon.
event: Event envelope encryption.
handler_ctx: Delivery metadata injectable into #[photon::subscribe] handlers (v2).
handler_descriptor: Handler descriptor for #[photon::subscribe] inventory registration.
handler_registry: Registry of #[photon::subscribe] handlers discovered via inventory.
host: Host-facing runtime wiring (executor bootstrap).
instrumentation: Photon platform observability via injected photon_telemetry::OpsLog.
models: In-memory models for Photon entities.
prelude: Common imports for application code using typed topics and the runtime handle.
registry: Topic registry for discovering and looking up registered topics.
runtime_parts: Host-facing runtime wiring helpers (Integrating the host).
shard_router: Virtual shard routing for consumer-group topics.
storage: Storage adapter port — one implementation per --storage tier (mem, sqlite, nats, fluvio, kafka).

Macros§

actor_downcast_methods: Re-export identity port traits and JSON stubs from photon_core. Expands the standard Actor Any downcast method bodies.

Structs§

AdminBackendSummary: Storage adapter capabilities surfaced for admin UIs.
AdminCheckpointSummary: Last committed sequence for a subscription/topic partition.
AdminHandlerSummary: Inventory entry for a #[photon::subscribe] handler.
AdminSnapshot: Point-in-time ops introspection snapshot (read-only).
AdminTopicSummary: Topic catalog entry from the compile-time TopicRegistry.
BackendCapabilities: Capabilities advertised by a storage port (surfaced via crate::backend::GenericPhotonBackend).
BackendContext: Shared builder context for storage adapter install fns.
ConsoleOpsLog: stderr/stdout structured lines (default dev adapter).
Envelope: Event envelope with decoded payload.
Event: Published event with payload and metadata.
FluvioConfig: Resolved Fluvio adapter settings (no env lookups at append time).
FluvioStoragePort: Fluvio-backed storage port.
FluvioStoragePortBuilder: Builder for super::port::FluvioStoragePort.
GenericPhotonBackend: Unified backend — all storage adapters install this wrapping their port.
GroupOpts: Options specific to consumer-group subscriptions.
HandlerCtx: Delivery metadata for a single handler invocation.
HandlerDescriptor: Descriptor for a #[photon::subscribe] handler (inventory + executor dispatch).
HandlerRegistry: Registry of inventory-submitted subscription handlers.
InProcStoragePort: In-process storage for the mem adapter tier.
JsonActor: Re-export identity port traits and JSON stubs from photon_core. Actor label derived from publish-time actor_json (System.operation when present).
JsonIdentityFactory: Re-export identity port traits and JSON stubs from photon_core. Reconstructs JsonActor from JSON captured at publish time.
KafkaConfig: Resolved Kafka adapter settings (no env lookups at append time).
KafkaStoragePort: Kafka-backed storage port.
KafkaStoragePortBuilder: Builder for super::port::KafkaStoragePort.
NatsConfig: Resolved NATS adapter settings (no env lookups at append time).
NatsStoragePort: NATS JetStream-backed storage port.
NatsStoragePortBuilder: Builder for super::port::NatsStoragePort.
NoOpsLog: Zero-cost no-op (benchmark telemetry=off and minimal CI).
Photon: Main Photon runtime handle.
PhotonBuilder: Builder for constructing Photon runtimes.
PhotonRuntimeState: Shared storage port, executor services, and handler dispatch controller.
ShardConfig: Virtual shard settings for DeliveryMode::ConsumerGroup topics.
SqliteStoragePort: Embedded SQLite storage for the sqlite adapter tier.
SubscribeOpts: Options for subscribing to a topic.
Subscription: Subscription handler registration.
SubscriptionHandle: Placeholder for future subscription lifecycle control (cancel, pause).
TopicDescriptor: Descriptor for a registered topic (from #[photon::topic] macro).
TopicMetadata: Topic metadata stored in the registry.
TopicRegistry: Registry of all topics discovered via #[photon::topic].
TransportCrypto: Symmetric payload encryption (XChaCha20-Poly1305).

Enums§

BrokerTransportSecurity: How Photon may connect to an external broker.
DeliveryMode: How publishes on a topic are routed to subscribers.
FluvioReplayCursor: How durable replay and checkpoints map to Fluvio offsets.
IdentityError: Re-export identity port traits and JSON stubs from photon_core. Errors from crate::IdentityFactory::reconstruct.
KafkaReplayCursor: How durable replay and checkpoints map to Kafka offsets.
PhotonError: Errors that can occur in Photon operations.
ReplayCursor: How durable replay and checkpoints map to JetStream.
SubscriptionMode: Subscription mode.

Constants§

ALLOW_INSECURE_BROKER_ENV: Environment opt-in for plaintext broker endpoints (development/CI only).
FLUVIO_ENDPOINT_ENV: Environment variable for Fluvio SC endpoint.
FLUVIO_MAX_INFLIGHT_ENV: Environment variable for max in-flight publishes per port.
FLUVIO_PREFIX_ENV: Environment variable for topic name prefix.
FLUVIO_REPLAY_CURSOR_ENV: Environment variable for replay cursor mode.
FLUVIO_REPLICAS_ENV: Environment variable for topic replication factor.
FLUVIO_RETENTION_ENV: Environment variable for Fluvio topic retention.
FLUVIO_SYNC_ACK_ENV: Environment variable for synchronous publish ack (1 / 0).
FLUVIO_TOPIC_SHARDS_ENV: Environment variable for topic shard count (bench/scripts fallback).
KAFKA_BROKERS_ENV: Environment variable for Kafka bootstrap brokers.
KAFKA_MAX_INFLIGHT_ENV: Environment variable for max in-flight publishes per port.
KAFKA_PREFIX_ENV: Environment variable for topic name prefix.
KAFKA_REPLAY_CURSOR_ENV: Environment variable for replay cursor mode.
KAFKA_REPLICAS_ENV: Environment variable for topic replication factor.
KAFKA_RETENTION_ENV: Environment variable for Kafka topic retention.
KAFKA_SYNC_ACK_ENV: Environment variable for synchronous publish ack (1 / 0).
KAFKA_TOPIC_SHARDS_ENV: Environment variable for topic shard count (bench/scripts fallback).
MAX_ERROR_MESSAGE_CHARS: Maximum length of persisted / logged handler error messages.
MAX_INFLIGHT_ENV: Environment variable for max in-flight publishes per port.
NATS_CREDS_ENV: Environment variable for NATS credentials file path (JWT + NKey .creds).
NATS_RETENTION_ENV: Environment variable for JetStream stream retention.
NATS_STREAM_ENV: Environment variable for JetStream stream name.
NATS_STREAM_SHARDS_ENV: Environment variable for stream shard count (bench/scripts fallback).
NATS_URL_ENV: Environment variable for NATS server URL.
REPLAY_CURSOR_ENV: Environment variable for replay cursor mode.
SQLITE_PATH_ENV: Environment variable for the SQLite database file path.
SYNC_ACK_ENV: Environment variable for synchronous publish ack (1 / 0).

Traits§

Actor: Re-export identity port traits and JSON stubs from photon_core. Opaque actor handle for handler execution.
IdentityFactory: Re-export identity port traits and JSON stubs from photon_core. Reconstruct handler identity from captured actor JSON (handler boundary only).
OpsLog: Structured ops metrics/events for publish, drain, DLQ, checkpoints.
PhotonBackend: Backend for publish/subscribe delivery and checkpoint persistence.
StoragePort: Storage adapter contract — append, subscribe, checkpoints, and optional retention.

Functions§

collect_admin_snapshot: Build a full admin snapshot for the given runtime (async checkpoint reads).
configure: Configure the default Photon instance used by macro-generated convenience helpers (Type::publish() / Type::subscribe()).
consumer_group_for: Stable group id alias (NATS deliver-subject analogue).
default: Clone of the process-wide Photon set by configure, if any.
durable_consumer_name: Build a Kafka-safe durable consumer group id for a keyed subscription.
fluvio_consumer_group_for: Stable group id alias (NATS deliver-subject analogue).
fluvio_durable_consumer_name: Build a Fluvio-safe durable consumer id for a keyed subscription.
fluvio_endpoint_from_env: Read Fluvio endpoint from the environment.
install_ops_log: Install the process-wide ops log (typically at server boot before Photon runtime).
kafka_brokers_from_env: Read Kafka brokers from the environment.
kafka_retention_ms: Convert Photon retention duration to Kafka retention.ms (milliseconds).
map_broker_connect_err: Broker connect failure labeled with a redacted endpoint and redacted source text.
ops_log: Resolved ops log — NoOpsLog until install_ops_log.
redact_credentials_in_text: Redact scheme://userinfo@host substrings embedded in free-form error text.
redact_endpoint: Remove URL userinfo before placing an endpoint in an error or log message.
resolve_photon_remote_base_url: Origin + path prefix before /api/photon (no trailing slash).
sanitize_error_message: Truncate and strip obvious secret-looking substrings from an error message.
sqlite_path_from_env: Resolve database path from PATH_ENV, or a unique file under the system temp dir.

Type Aliases§

EmbeddedBackend: Back-compat alias for the default in-process mem tier GenericPhotonBackend.
Result: Result type alias for Photon operations.

Attribute Macros§

subscribe: Register an async handler (#[photon::subscribe]). Marks a function as a subscription handler registered via inventory.
topic: Register a typed topic struct (#[photon::topic]). Marks a struct as a Photon topic, generating typed publish/subscribe APIs.

Crate photon

Crate photon Copy item path

§Features

§Getting started

§Choose a topology

§Embedded (one binary)

§Brokered (publisher + worker binaries)

§What you create

§Shared setup (both binaries)

§Publisher binary

§Worker binary

§Run both

§3. Declare topics and handlers

§4. Publish and subscribe

§5. Run the executor and reclaim

§Notes

§Architecture

Re-exports§

Modules§

Macros§

Structs§

Enums§

Constants§

Traits§

Functions§

Type Aliases§

Attribute Macros§

Crate photon