BE-942: Report integration progress and workspace metrics

libs/@local/integrations/engine/Cargo.toml

@@ -20,13 +20,15 @@ duckdb = { workspace = true, public = true, features = ["bundled", "json"] }
 
 # Private third-party dependencies
 derive_more    = { workspace = true, features = ["display", "error"] }
+opentelemetry  = { workspace = true, features = ["metrics"] }
 serde_json     = { workspace = true }
 simple-mermaid = { workspace = true }
 tokio          = { workspace = true, features = ["sync"] }
 tracing        = { workspace = true, features = ["attributes"] }
 uuid           = { workspace = true, features = ["v5"] }
 
 [dev-dependencies]
+opentelemetry_sdk = { workspace = true, features = ["metrics", "testing"] }
 pretty_assertions = { workspace = true }
 proptest          = { workspace = true, features = ["attr-macro", "std"] }
 serde             = { workspace = true, features = ["derive"] }

libs/@local/integrations/engine/README.md

@@ -5,3 +5,17 @@ Runs an integration's pipelines in DuckDB.
 Each run gets a `Workspace`, a DuckDB database file in which the run's queries execute. A query is a single `SELECT`, and cannot open files, load extensions or change settings.
 
 `derive_entity_id` derives stable entity IDs from an integration namespace, entity type and source record ID using UUID v5.
+
+`Progress` counts an operation's observed work and emits structured tracing events at most every ten seconds as work advances. Call `report` for a final observation, including short operations. Reports carry the operation name, observed count, optional expected total, and elapsed seconds in the tracing span where tracking started. `snapshot` reads the counters for other reporting backends.
+
+Workspace operations export OpenTelemetry metrics through the application's global meter provider. Initialize it before opening workspaces, for example through `hash_telemetry::TelemetryRegistry::init_global`.
+
+| Metric                                  | Observed value                                                                                     |
+| --------------------------------------- | -------------------------------------------------------------------------------------------------- |
+| `integrations.workspace.requests`       | Request counts by operation and outcome: `success`, `failure`, `cancelled`, or `skipped`.          |
+| `integrations.workspace.duration`       | Request execution duration in seconds, excluding queue time.                                       |
+| `integrations.workspace.rows`           | Rows read by queries or materialized into tables, including work followed by a disk-limit failure. |
+| `integrations.workspace.disk.usage`     | Workspace bytes measured by existing disk-limit checks.                                            |
+| `integrations.workspace.snapshot.bytes` | Bytes copied to successfully published snapshots.                                                  |
+
+The operation labels are `open`, `query`, `create_table`, `snapshot`, and `close`. `cancelled` means a caller dropped an executing request; `skipped` means it dropped the request before execution. Metrics aggregate across workspaces.

libs/@local/integrations/engine/src/lib.rs

@@ -17,13 +17,15 @@ extern crate alloc;
 
 mod byte_size;
 mod entity_id;
+mod progress;
 mod sql;
 mod variant_name;
 mod workspace;
 
 pub use self::{
     byte_size::ByteSize,
     entity_id::derive_entity_id,
+    progress::{Progress, ProgressSnapshot},
     workspace::{
         CloseError, DataErrorType, DiskUsageError, OpenError, QueryErrorType, Rows, Setting,
         SnapshotError, StatementError, Workspace, WorkspaceConfig,

libs/@local/integrations/engine/src/progress/mod.rs

@@ -0,0 +1,100 @@
+#[cfg(test)]
+mod tests;
+
+use core::{
+    sync::atomic::{AtomicU64, Ordering},
+    time::Duration,
+};
+use std::time::Instant;
+
+use tracing::Span;
+
+const REPORT_INTERVAL: Duration = Duration::from_secs(10);
+
+/// Tracks an operation's observed work and reports it through structured tracing events.
+///
+/// Share it between tasks to accumulate completed work. Periodic reports are emitted by
+/// [`advance`](Self::advance); call [`report`](Self::report) for a final observation. Events
+/// belong to the tracing span in which the operation started.
+#[derive(Debug)]
+pub struct Progress {
+    operation: Box<str>,
+    total: Option<u64>,
+    completed: AtomicU64,
+    started: Instant,
+    last_report_ms: AtomicU64,
+    span: Span,
+}
+
+/// The observed work and elapsed time of an operation at the time of a report.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ProgressSnapshot {
+    /// The number of work items observed so far.
+    pub completed: u64,
+    /// The expected number of work items, if known when the operation started.
+    pub total: Option<u64>,
+    /// The time since the operation started.
+    pub elapsed: Duration,
+}
+
+impl Progress {
+    /// Starts tracking an operation, with an optional expected work count.
+    #[must_use]
+    pub fn start(operation: impl Into<Box<str>>, total: Option<u64>) -> Self {
+        Self {
+            operation: operation.into(),
+            total,
+            completed: AtomicU64::new(0),
+            started: Instant::now(),
+            last_report_ms: AtomicU64::new(0),
+            span: Span::current(),
+        }
+    }
+
+    /// Adds observed work, saturating at the counter's maximum, and reports at most every 10
+    /// seconds.
+    pub fn advance(&self, count: u64) {
+        self.completed
+            .update(Ordering::Relaxed, Ordering::Relaxed, |completed| {
+                completed.saturating_add(count)
+            });
+        if self.claim_report(self.started.elapsed()) {
+            self.report();
+        }
+    }
+
+    /// Reads the observed work count and elapsed time.
+    #[must_use]
+    pub fn snapshot(&self) -> ProgressSnapshot {
+        ProgressSnapshot {
+            completed: self.completed.load(Ordering::Relaxed),
+            total: self.total,
+            elapsed: self.started.elapsed(),
+        }
+    }
+
+    /// Emits the observed progress immediately, including operations shorter than the report
+    /// interval.
+    pub fn report(&self) {
+        let snapshot = self.snapshot();
+        tracing::info!(
+            parent: &self.span,
+            operation = %self.operation,
+            completed = snapshot.completed,
+            total = snapshot.total,
+            elapsed_seconds = snapshot.elapsed.as_secs_f64(),
+            "integration progress"
+        );
+    }
+
+    /// Reserves a periodic report when its interval has elapsed.
+    fn claim_report(&self, elapsed: Duration) -> bool {
+        let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
+        self.last_report_ms
+            .try_update(Ordering::Relaxed, Ordering::Relaxed, |last| {
+                (Duration::from_millis(elapsed_ms.saturating_sub(last)) >= REPORT_INTERVAL)
+                    .then_some(elapsed_ms)
+            })
+            .is_ok()
+    }
+}

libs/@local/integrations/engine/src/progress/tests.rs

@@ -0,0 +1,102 @@
+use core::time::Duration;
+use std::thread;
+
+use pretty_assertions::assert_eq;
+
+use super::Progress;
+
+#[test]
+fn progress_concurrent_updates() {
+    let progress = Progress::start("apply", Some(800));
+    thread::scope(|scope| {
+        for _ in 0..8 {
+            scope.spawn(|| {
+                for _ in 0..100 {
+                    progress.advance(1);
+                }
+            });
+        }
+    });
+
+    let snapshot = progress.snapshot();
+    assert_eq!(
+        snapshot.completed, 800,
+        "all observed work should be counted"
+    );
+    assert_eq!(
+        snapshot.total,
+        Some(800),
+        "the expected total should remain fixed"
+    );
+}
+
+#[test]
+fn progress_exceeds_expected_total() {
+    let progress = Progress::start("apply", Some(2));
+    progress.advance(3);
+    assert_eq!(
+        progress.snapshot().completed,
+        3,
+        "observed work should not be capped by an estimate"
+    );
+
+    let progress = Progress::start("scan", None);
+    progress.advance(u64::MAX);
+    progress.advance(1);
+    let snapshot = progress.snapshot();
+    assert_eq!(
+        snapshot.completed,
+        u64::MAX,
+        "the counter should saturate instead of wrapping"
+    );
+    assert_eq!(
+        snapshot.total, None,
+        "an unknown total should remain unknown"
+    );
+}
+
+#[test]
+fn progress_report_interval() {
+    let progress = Progress::start("apply", None);
+    assert!(
+        !progress.claim_report(Duration::from_millis(9_999)),
+        "an early report should be suppressed"
+    );
+    assert!(
+        progress.claim_report(Duration::from_secs(10)),
+        "the first report should be due after 10 seconds"
+    );
+    assert!(
+        !progress.claim_report(Duration::from_secs(10)),
+        "only one task should claim each report"
+    );
+    assert!(
+        !progress.claim_report(Duration::from_secs(9)),
+        "an older observation should not move the interval backwards"
+    );
+    assert!(
+        progress.claim_report(Duration::from_secs(20)),
+        "the next interval should allow another report"
+    );
+}
+
+#[test]
+fn progress_concurrent_reports() {
+    let progress = Progress::start("apply", None);
+    let reports = thread::scope(|scope| {
+        let workers: Vec<_> = core::iter::repeat_with(|| {
+            scope.spawn(|| progress.claim_report(Duration::from_secs(10)))
+        })
+        .take(8)
+        .collect();
+        workers
+            .into_iter()
+            .map(|worker| worker.join().expect("reporting thread should finish"))
+            .filter(|claimed| *claimed)
+            .count()
+    });
+    assert_eq!(
+        reports, 1,
+        "concurrent updates should reserve only one periodic report"
+    );
+}