fix: cancel media uploads when mask_otel_spans drops the batch

[uuzzrm]

What does this PR do?

Fixes #16102

EventSerializer handles media extraction before the mask_otel_spans hook runs, and that ordering is deliberate (the hook must see @@@langfuseMedia: reference strings, not raw bytes). The problem is the consequence: when the hook raises or returns an invalid result, the whole batch is dropped as documented, but the media upload jobs were already enqueued during attribute processing and still get uploaded. The fail-closed guarantee that "dropping the batch" covers the spans does not extend to the media behind them.

This tracks the media IDs enqueued while processing a batch's attributes and cancels them when the mask hook drops the batch. The upload consumer skips cancelled media IDs, so a redacted attribute never produces an upload.

Type of change

• Bug fix

Verification

pytest tests/unit/test_mask_otel_spans.py tests/unit/test_serializer.py -q -k "not path"   # 81 passed
pytest tests/unit/test_otel.py tests/unit/test_media.py -q                                  # 85 passed, 2 skipped
ruff check langfuse/_client/span_exporter.py langfuse/_task_manager/media_manager.py tests/unit/test_mask_otel_spans.py
ruff format --check <same files>
mypy langfuse/_client/span_exporter.py langfuse/_task_manager/media_manager.py --no-error-summary

The new regression test test_mask_otel_spans_drop_batch_cancels_enqueued_media_uploads fails without the fix (the upload job is processed despite the batch being dropped) and passes with it. test_path fails on this machine only (POSIX-vs-Windows path separators) and passes in CI's Linux runner.

Checklist

• I self-reviewed the diff using code_review.md.
• I added or updated tests for behavior changes.
• I updated docs, examples, or .env.template if needed.
• I did not hand-edit generated files; if generated files changed, I used the upstream regeneration path.
• I did not commit secrets or credentials.

Greptile Summary

This PR attempts to extend fail-closed OpenTelemetry span masking to separately queued media uploads.

• Tracks media IDs produced while transforming an export batch.
• Marks those IDs as cancelled when the masking hook drops the batch.
• Makes the media consumer skip jobs whose IDs are marked as cancelled.
• Adds a sequential regression test for the cancellation path.

Confidence Score: 2/5

This PR is not yet safe to merge because dropped spans can still upload media under normal consumer concurrency, and content-based cancellation can suppress or misapply uploads across batches.

Media jobs are visible to background consumers before masking completes, and the sole pre-upload cancellation check can be passed before cancellation is recorded; additionally, deterministic media IDs cannot distinguish individual queued jobs from different batches.

Files Needing Attention: langfuse/_task_manager/media_manager.py, langfuse/_client/span_exporter.py, tests/unit/test_mask_otel_spans.py

Security Review

The new cancellation check does not close the concurrent upload window: a background consumer can dequeue and begin uploading media before the masking failure records the cancellation, allowing media from a dropped span to leave the process.

Sequence Diagram

sequenceDiagram
    participant E as Export thread
    participant Q as Media queue
    participant C as Media consumer
    participant M as Mask hook
    participant B as Backend
    E->>Q: Enqueue extracted media
    Q-->>C: Dequeue job
    C->>C: Check cancellation marker
    E->>M: Invoke mask_otel_spans
    C->>B: Begin media upload
    M-->>E: Raise / invalid result
    E->>C: Record cancelled media ID
    Note over C,B: Cancellation arrives after the only check

Loading

Prompt To Fix All With AI

### Issue 1
langfuse/_task_manager/media_manager.py:69-75
**Cancellation misses active uploads**

When a media consumer dequeues a job before or while `mask_otel_spans` runs, it can pass this sole cancellation check and begin the network upload before the hook drops the batch, causing media from a redacted span to be uploaded. **How this was verified:** Media is enqueued before the synchronous mask hook runs, while independent consumer threads dequeue jobs and never re-check cancellation inside the upload path.

### Issue 2
langfuse/_task_manager/media_manager.py:69-70
**Content IDs misroute cancellation**

When an exported batch and a dropped batch enqueue identical media bytes, both jobs have the same deterministic `media_id`, so this one-shot marker can skip the earlier legitimate job and then allow the dropped job to upload. A stale marker can likewise suppress a later legitimate upload of the same content.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

_{Reviews (1): Last reviewed commit: "fix: cancel media uploads when mask_otel..." | Re-trigger Greptile}

Greptile also left 2 inline comments on this PR.

Context used:

• Knowledge Base — OTel Span Processing and Export Pipeline
• Knowledge Base — Task Manager: Media Upload and Score Ingestion

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[greptile-apps]

Cancellation misses active uploads

When a media consumer dequeues a job before or while mask_otel_spans runs, it can pass this sole cancellation check and begin the network upload before the hook drops the batch, causing media from a redacted span to be uploaded. How this was verified: Media is enqueued before the synchronous mask hook runs, while independent consumer threads dequeue jobs and never re-check cancellation inside the upload path.

Knowledge Base Used:

• OTel Span Processing and Export Pipeline
• Task Manager: Media Upload and Score Ingestion

Prompt To Fix With AI

This is a comment left during a code review.
Path: langfuse/_task_manager/media_manager.py
Line: 69-75

Comment:
**Cancellation misses active uploads**

When a media consumer dequeues a job before or while `mask_otel_spans` runs, it can pass this sole cancellation check and begin the network upload before the hook drops the batch, causing media from a redacted span to be uploaded. **How this was verified:** Media is enqueued before the synchronous mask hook runs, while independent consumer threads dequeue jobs and never re-check cancellation inside the upload path.

**Knowledge Base Used:**
- [OTel Span Processing and Export Pipeline](https://app.greptile.com/personal-org-4986/-/custom-context/knowledge-base/langfuse/langfuse-python/-/docs/otel-pipeline.md)
- [Task Manager: Media Upload and Score Ingestion](https://app.greptile.com/personal-org-4986/-/custom-context/knowledge-base/langfuse/langfuse-python/-/docs/task-manager.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

[greptile-apps]

Content IDs misroute cancellation

When an exported batch and a dropped batch enqueue identical media bytes, both jobs have the same deterministic media_id, so this one-shot marker can skip the earlier legitimate job and then allow the dropped job to upload. A stale marker can likewise suppress a later legitimate upload of the same content.

Knowledge Base Used: Task Manager: Media Upload and Score Ingestion

Prompt To Fix With AI

This is a comment left during a code review.
Path: langfuse/_task_manager/media_manager.py
Line: 69-70

Comment:
**Content IDs misroute cancellation**

When an exported batch and a dropped batch enqueue identical media bytes, both jobs have the same deterministic `media_id`, so this one-shot marker can skip the earlier legitimate job and then allow the dropped job to upload. A stale marker can likewise suppress a later legitimate upload of the same content.

**Knowledge Base Used:** [Task Manager: Media Upload and Score Ingestion](https://app.greptile.com/personal-org-4986/-/custom-context/knowledge-base/langfuse/langfuse-python/-/docs/task-manager.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.