ifc_lite_server/services/

streaming.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! Streaming geometry processing with Server-Sent Events.
//!
//! Thin bridge over the canonical `ifc_lite_processing` pipeline: the
//! blocking task runs `process_geometry_streaming_filtered_with_baked_basis`
//! (`..._with_options` plus the frame published before the first batch,
//! #5407; the same code path as `POST /api/v1/parse` and the wasm
//! `processGeometryBatch` boundary) and forwards its batch callbacks through
//! a bounded channel as [`StreamEvent`]s (see [`EVENT_BUFFER_EVENTS`]).
//!
//! This file used to host a third, bespoke geometry pipeline with its own
//! scan, style index (SurfaceColour-only, no material chain, no indexed
//! colour maps), no aggregate void propagation, no submeshes and no type
//! geometry — meshes streamed from `/parse/stream` could differ from every
//! other surface (alignment audit). Supersede means delete: it is gone, and
//! the streaming endpoints inherit every pipeline feature (and bug fix)
//! automatically, including `opening_filter` support which the bespoke
//! pipeline never had.

use crate::services::cache::DiskCache;
use crate::types::{finish_meshes, StreamEvent};
use ifc_lite_processing::style::ModelFinishes;
use async_stream::stream;
use futures::Stream;
use ifc_lite_processing::{
    extract_symbolic_data_with_provenance_in_frame, process_geometry_streaming_filtered_with_baked_basis, OpeningFilterMode,
    StreamingOptions, TessellationQuality,
};
use std::pin::Pin;
use tokio::sync::mpsc;

/// How many events the blocking producer may run ahead of the response
/// stream before it waits for the client.
///
/// The channel between the two is the only place the parse's emit rate meets
/// the client's read rate. A client that stops reading keeps its connection
/// until the write-idle timeout closes it (`IFC_STREAM_IDLE_TIMEOUT_SECS`,
/// see `write_timeout.rs`); until then only this bound stops the producer
/// from queuing every batch of the model on top of the parse's working set,
/// which admission does not account for (it charges the UPLOAD size, not the
/// tessellated output). A full queue parks the producer in `blocking_send`
/// (legal there, it runs on a `spawn_blocking` thread), so a stalled reader
/// stalls its own parse instead of growing the process.
///
/// Four events is two batches: the pipeline emits each batch as a `Batch`
/// frame followed by its `Progress` frame, so the consumer always has one
/// batch to serialise and one queued behind it while the producer meshes the
/// next. Nothing is dropped: the producer waits for capacity. A receiver that
/// goes away makes a parked send return at once, and the next callback sees
/// `tx.is_closed()` and cancels.
pub(crate) const EVENT_BUFFER_EVENTS: usize = 4;

fn find_bytes(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack
        .windows(needle.len())
        .position(|window| window == needle)
}

/// Detect the declared IFC schema from the STEP header.
///
/// Schema-like text in DATA values or comments must not influence metadata.
pub(crate) fn detect_schema_version(content: &[u8]) -> &'static str {
    let header_end = find_bytes(content, b"ENDSEC;").unwrap_or(content.len());
    let header = &content[..header_end];
    let Some(schema_start) = find_bytes(header, b"FILE_SCHEMA") else {
        return "IFC2X3";
    };
    let declaration = &header[schema_start..];
    let declaration_end = declaration
        .iter()
        .position(|byte| *byte == b';')
        .unwrap_or(declaration.len());
    let declaration = &declaration[..declaration_end];

    if find_bytes(declaration, b"IFC4X3").is_some() {
        "IFC4X3"
    } else if find_bytes(declaration, b"IFC4").is_some() {
        "IFC4"
    } else {
        "IFC2X3"
    }
}

/// Generate streaming geometry events backed by the canonical pipeline.
///
/// Takes the raw IFC bytes (issue #1023): localized non-UTF-8 byte sequences
/// in the HEADER must not block otherwise valid models, so no `String`
/// conversion happens anywhere on this path.
pub fn process_streaming(
    content: bytes::Bytes,
    initial_batch_size: usize,
    max_batch_size: usize,
    opening_filter: OpeningFilterMode,
    tessellation_quality: TessellationQuality,
    admission: Option<crate::admission::AdmissionGuard>,
) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
    // Zero is a caller bug, not a reason to stall the stream.
    let initial_batch_size = initial_batch_size.max(1);
    let max_batch_size = max_batch_size.max(1);

    let (tx, mut rx) = mpsc::channel::<StreamEvent>(EVENT_BUFFER_EVENTS);

    // Disconnect-aware cancellation: when the SSE client hangs up, the
    // receiver drops, the batch callback notices via `tx.is_closed()`, and the
    // streaming core stops between chunks instead of meshing the rest of the
    // model for nobody (a full-size parse used to keep burning a core and its
    // memory slot to completion).
    let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let cancel_for_task = std::sync::Arc::clone(&cancel);

    // The admission permit must be held until BOTH sides are done: the
    // blocking producer (on disconnect the stream drops first, but the task
    // keeps its memory/CPU until the cooperative cancel takes effect) AND the
    // response stream (the channel can still hold `EVENT_BUFFER_EVENTS`
    // emitted events after the producer exits, so dropping the permit at task
    // exit would let a replacement parse be admitted on top of the undrained
    // buffers). An Arc'd guard held by both releases on whichever finishes
    // last.
    let admission = admission.map(std::sync::Arc::new);
    let admission_for_task = admission.clone();

    let handle = tokio::task::spawn_blocking(move || {
        let _admission = admission_for_task;
        let cache_key = DiskCache::generate_key(&content);

        let mut started = false;
        let mut batch_number = 0usize;
        let mut last_type = String::new();
        // Filled by the pipeline once it has chosen the frame, before the
        // first batch (#5407).
        let baked_basis = std::sync::OnceLock::new();
        // #5984: every batch's meshes carry their IFC-authored finish. One
        // styled-item scan up front; a file that authors none joins nothing.
        let mut finishes = ModelFinishes::from_content(&content);

        let result = process_geometry_streaming_filtered_with_baked_basis(
            &content,
            opening_filter,
            StreamingOptions {
                initial_batch_size,
                throughput_batch_size: max_batch_size,
                tessellation_quality,
                // Batches are forwarded as they are emitted — retaining them
                // in the ProcessingResult would double peak memory.
                retain_emitted_meshes: false,
                cancel: Some(std::sync::Arc::clone(&cancel_for_task)),
                ..StreamingOptions::default()
            },
            &baked_basis,
            |meshes, processed, total| {
                if tx.is_closed() {
                    cancel_for_task.store(true, std::sync::atomic::Ordering::Relaxed);
                    return;
                }
                if !started {
                    started = true;
                    let _ = tx.blocking_send(StreamEvent::Start {
                        total_estimate: total,
                    });
                    let _ = tx.blocking_send(StreamEvent::Progress {
                        processed: 0,
                        total,
                        current_type: "indexing".into(),
                    });
                }
                if let Some(mesh) = meshes.last() {
                    last_type = mesh.ifc_type.clone();
                }
                if !meshes.is_empty() {
                    batch_number += 1;
                    let _ = tx.blocking_send(StreamEvent::Batch {
                        meshes: finish_meshes(&mut finishes, meshes.to_vec()),
                        batch_number,
                        baked_basis: baked_basis.get().copied(),
                    });
                }
                let _ = tx.blocking_send(StreamEvent::Progress {
                    processed,
                    total,
                    current_type: last_type.clone(),
                });
            },
            // Styling is eager on this path (`fast_first_batch` defaults to
            // false), so colour updates never fire.
            |_| {},
            |_| {},
        );

        if cancel_for_task.load(std::sync::atomic::Ordering::Relaxed) || tx.is_closed() {
            // Client gone: the partial result must not be presented as a
            // completed parse - skip Complete AND the symbolic extraction
            // (which re-scans the file). The is_closed check also covers a
            // disconnect after the LAST batch, where the callback can no
            // longer observe it.
            tracing::info!("SSE client disconnected; streaming parse stopped early");
            return;
        }

        if !started {
            // Zero-geometry model: the batch callback never ran. Emit Start
            // so consumers still observe the Start → Complete contract.
            let _ = tx.blocking_send(StreamEvent::Start { total_estimate: 0 });
        }

        // 2D symbolic stream (IfcAnnotation + IfcGrid) on the same blocking
        // thread — parity with the synchronous endpoints (issue #900).
        // Georeferencing already rides in `result.metadata`.
        //
        // In the frame the batches above were meshed in: the `Complete` event
        // carries the symbols and `mesh_coordinate_space` together, so they
        // must agree (#4706).
        let symbolic_data =
            extract_symbolic_data_with_provenance_in_frame(&content, result.frame);

        let _ = tx.blocking_send(StreamEvent::Complete {
            stats: result.stats,
            metadata: result.metadata,
            cache_key,
            mesh_coordinate_space: Some(result.mesh_coordinate_space),
            site_transform: result.site_transform,
            building_transform: result.building_transform,
            symbolic_data,
        });
        // `tx` drops here, closing the channel and ending the stream below.
    });

    Box::pin(stream! {
        let _admission = admission;
        while let Some(event) = rx.recv().await {
            yield event;
        }
        // Surface a panicked/cancelled blocking task as a stream error
        // instead of silently truncating the SSE stream.
        if let Err(e) = handle.await {
            yield StreamEvent::Error {
                message: format!("Streaming geometry task failed: {e}"),
            };
        }
    })
}

#[cfg(test)]
#[path = "streaming_tests.rs"]
mod streaming_tests;