Struct WebIngestionService 

pub struct WebIngestionService { /* private fields */ }

Implementations

impl WebIngestionService

pub fn new( lake: DataLakeConnection, default_audience: WriteAudience, ) -> WebIngestionService

pub fn default_audience(&self) -> &WriteAudience

The deployment’s resolved default audience – see the default_audience field’s doc comment. Exposed so the HTTP edge (resolve_write_audience) and the five OTLP IdentityContext construction sites can reach it without re-resolving the environment.

pub async fn check_ready(&self) -> bool

pub async fn from_env( default_audience: WriteAudience, ) -> Result<Arc<WebIngestionService>, Error>

Reads MICROMEGAS_SQL_CONNECTION_STRING and MICROMEGAS_OBJECT_STORE_URI, connects to the data lake, runs ingestion migrations, and returns a ready-to-use service.

default_audience – the deployment’s resolved default audience – is threaded straight into Self::new. micromegas-ingestion depends on neither micromegas-auth nor micromegas-analytics, so it cannot resolve this itself; the caller resolves it (typically via micromegas_auth::policy::default_audience_from_env) and passes it in.

pub fn insert_block( &self, body: Bytes, audience: &WriteAudience, ) -> impl Future<Output = Result<(), IngestionServiceError>>

pub fn insert_block_typed( &self, block: Block, audience: &WriteAudience, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Inserts a block whose payload is already typed (no envelope round-trip on the caller side).

The caller hands us a fully-built Block; we CBOR-encode the payload envelope once, write it to object storage, and INSERT the row. Used by the OTLP adapter where constructing the CBOR Block envelope just so insert_block could decode it would be wasted work.

Stamps the row with audience: the block’s own column, not derived from the process_id/stream_id it claims. A credential bound to audience A that knows another audience’s process_id/stream_id can still append a row naming those ids, but the row is labelled A, so blocks_view’s mismatch predicate excludes it from every reader’s view rather than surfacing it mislabelled to either audience.

pub fn insert_stream( &self, body: Bytes, audience: &WriteAudience, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a stream whose blocks will be ingested in the transit format, stamping it with audience – resolved by the caller from the authenticated credential, exactly as insert_process resolves its own.

This accepts any process_id unconditionally; there is no check that the caller is authorized to write a stream onto that process. The stream is stamped with the caller’s own audience, not the process’s, so blocks_view’s mismatch predicate is what keeps a later block from surfacing under the wrong audience if process_id belongs to someone else.

A re-registration of an existing stream_id under a different audience is rejected via Self::check_stream_audience_conflict, mirroring Self::check_process_audience_conflict.

pub fn register_otel_stream( &self, stream_id: Uuid, process_id: Uuid, tags: Vec<String>, format: &str, audience: &WriteAudience, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a stream produced by an OTLP ingestion path, stamping it with audience, exactly like Self::insert_stream.

dependencies_metadata and objects_metadata are filled with the CBOR sentinel for an empty Vec<UserDefinedType> so legacy decode sites continue to work. format distinguishes per-block dispatch downstream (e.g. "otlp/v1/logs"). Stream properties are always empty for OTel — scope and per-event attrs live on individual rows during materialization, not on the stream.

Hack: piggybacking OTLP onto the transit-shaped streams row (with empty metadata sentinels) is expedient for two formats but won’t scale. To support more formats cleanly, dependencies_metadata, objects_metadata, and format should be merged into a single per-format payload column.

Same conflict guard as Self::insert_stream – see Self::check_stream_audience_conflict.

pub fn insert_process( &self, body: Bytes, audience: &WriteAudience, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a process from the native (CBOR) ingestion path, stamping it with audience. audience is resolved by the caller from the authenticated credential (AuthContext.bound_audience); this method never trusts a client-supplied micromegas.* property – strip_reserved_properties drops it, and there is no property to re-append: the stamp lands on the audience column instead.

A conflicting re-registration (an existing process_id under a different audience than audience) is rejected with IngestionServiceError::AudienceConflict rather than silently no-op’d: process_id is client-chosen on this path, so a reused id under a different credential is a real, reachable case, and it is what keeps ownership_rewrite.rs’s MAX(audience) per-process resolution sound. An existing row with a NULL audience column (legacy data with no stamp; admin replication hard-fails on a missing audience column) is resolved to the deployment default the same way every reader does, then compared like any other row; it is never retro-stamped, ever – see the module-level design doc.

pub fn register_otel_process( &self, process_id: Uuid, exe: String, username: String, computer: String, distro: String, cpu_brand: String, tsc_frequency: i64, start_time: DateTime<Utc>, start_ticks: i64, properties: Vec<Property>, audience: &WriteAudience, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a process originating from OTLP. Idempotent via ON CONFLICT DO NOTHING.

realname is set equal to username (OTel has no separate “real name” concept). parent_process_id is always NULL — OTel has no parent-process model. insert_time is the server wall clock, matching the existing insert_process path.

audience is stamped on the audience column, exactly like insert_process; client-supplied micromegas.* properties are stripped via strip_reserved_properties the same way. Same conflict guard as insert_process, and for a confidentiality reason, not just consistency: processes is a single table shared with the native path, and insert_process accepts a client-chosen process_id stamped with the caller’s own audience. Because process_id_from_resource’s derivation formula is public, any ingestion credential can pre-register (via the native path) the exact process_id a victim audience’s OTLP producer will later derive; without this guard the genuine producer’s stream/blocks would silently land on a row stamped with the squatter’s audience, leaking that audience’s data to the squatter. check_process_audience_conflict closes that hole the same way it does for insert_process.

Trait Implementations

impl Clone for WebIngestionService

fn clone(&self) -> WebIngestionService

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.