[ENH](fn-consumer): Add concurrency gauge

[tanujnay112]

Summary

• add an unlabeled fn_consumer_current_compactions observable gauge
• track dispatch futures only while they are actively running
• decrement safely on success, failure, panic, or task cancellation

Testing

• not run at request; formatting and commit hooks passed

[claude]

Claude Code Review

This repository is configured for manual code reviews. Comment @claude review for a one-time review, or @claude review always to subscribe this PR to a review on every future push.

_{Tip: disable this comment in your organization's Code Review settings.}

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-09-02T00:05:36.754645Z	76878fc	PR opened

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[github-actions]

Reviewer Checklist

Please leverage this checklist to ensure your code review is thorough before approving

Testing, Bugs, Errors, Logs, Documentation

• Can you think of any use case in which the code does not behave as intended? Have they been tested?
• Can you think of any inputs or external events that could break the code? Is user input validated and safe? Have they been tested?
• If appropriate, are there adequate property based tests?
• If appropriate, are there adequate unit tests?
• Should any logging, debugging, tracing information be added or removed?
• Are error messages user-friendly?
• Have all documentation changes needed been made?
• Have all non-obvious changes been commented?

System Compatibility

• Are there any potential impacts on other parts of the system or backward compatibility?
• Does this change intersect with any items on our roadmap, and if so, is there a plan for fitting them together?

Quality

• Is this code of a unexpectedly high quality (Readability, Modularity, Intuitiveness)

[graphite-app]

The observable gauge instrument returned by .build() is being dropped immediately, which will likely cause the callback to stop being invoked and the metric to not be reported.

OpenTelemetry observable instruments typically need to be kept alive for their callbacks to continue working. The return value should be stored in the FnConsumerMetrics struct:

struct FnConsumerMetrics {
    current_compactions: Arc<AtomicU64>,
    _gauge: ObservableGauge<u64>, // Keep instrument alive
}

impl Default for FnConsumerMetrics {
    fn default() -> Self {
        let current_compactions = Arc::new(AtomicU64::new(0));
        let observed_count = current_compactions.clone();
        let gauge = opentelemetry::global::meter("chroma_fn_consumer")
            .u64_observable_gauge("fn_consumer_current_compactions")
            .with_description("Number of compaction jobs currently running in fn-consumer")
            .with_callback(move |observer| {
                observer.observe(observed_count.load(Ordering::Relaxed), &[]);
            })
            .build();
        Self {
            current_compactions,
            _gauge: gauge,
        }
    }
}

Without storing the instrument, the metric collection will not work in production.

Spotted by Graphite



Is this helpful? React 👍 or 👎 to let us know.

[tanujnay112]

I checked the OpenTelemetry Rust 0.27 implementation used by this repo. During .build(), the SDK passes the callback to Pipelines::register_callback, which stores it as an Arc in PipelineInner.callbacks. The returned ObservableGauge is only a PhantomData marker and has no Drop behavior, so dropping that value does not unregister the callback or stop collection. The pipeline continues to own the callback and the captured atomic value, so this metric remains live.

[rescrv]

I believe that dropping this may lose the metrics. Otherwise aI don't get the reason for the clone, either. It reads funny.

[tanujnay112]

The extra clone is indeed unnecessary and reads oddly: the original metrics value is dropped immediately, so it could be moved directly into the spawned task. That is a harmless readability issue rather than a correctness issue. I also checked the gauge lifetime against OpenTelemetry Rust 0.27: .build() registers the callback into the SDK pipeline, which owns it independently; the returned ObservableGauge is a marker with no Drop behavior, so dropping it does not lose the metric.