feat(ws): improve warehouse-sources-load observability

[estefaniarabadan]

Problem

we need more logs to understand what's happening under pipelines V3

Changes

• add logs on pipeline V3 (both producer and consumer)

How did you test this code?

Test pass

Publish to changelog?

NO

[greptile-apps]

Prompt To Fix All With AI

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
posthog/temporal/data_imports/pipelines/pipeline_v3/load/processor.py:140
Using `capture_exception` with a freshly-constructed `Exception` is a misuse of the error-tracking API. Sentry treats this as an exception event: it groups all occurrences under the same issue (no real traceback, no surrounding context), and the grouping algorithm will conflate different counts as one noisy fingerprint. The situation has already been logged at `WARNING` level by structlog immediately above — if Sentry alerting is desired, `capture_message` is the appropriate call here since there is no real exception to capture.

```suggestion
            capture_message(f"Found {len(modified_files)} modified delta files during first sync")
```

_{Reviews (1): Last reviewed commit: "add batch index" | Re-trigger Greptile}

[greptile-apps]

Using capture_exception with a freshly-constructed Exception is a misuse of the error-tracking API. Sentry treats this as an exception event: it groups all occurrences under the same issue (no real traceback, no surrounding context), and the grouping algorithm will conflate different counts as one noisy fingerprint. The situation has already been logged at WARNING level by structlog immediately above — if Sentry alerting is desired, capture_message is the appropriate call here since there is no real exception to capture.

Suggested change

	)
	capture_message(f"Found {len(modified_files)} modified delta files during first sync")

Prompt To Fix With AI

This is a comment left during a code review.
Path: posthog/temporal/data_imports/pipelines/pipeline_v3/load/processor.py
Line: 140

Comment:
Using `capture_exception` with a freshly-constructed `Exception` is a misuse of the error-tracking API. Sentry treats this as an exception event: it groups all occurrences under the same issue (no real traceback, no surrounding context), and the grouping algorithm will conflate different counts as one noisy fingerprint. The situation has already been logged at `WARNING` level by structlog immediately above — if Sentry alerting is desired, `capture_message` is the appropriate call here since there is no real exception to capture.

```suggestion
            capture_message(f"Found {len(modified_files)} modified delta files during first sync")
```

How can I resolve this? If you propose a fix, please make it concise.

[danielcarletti]

Sentry? I though that was phased out. Does this make sense for us?

[danielcarletti]

LOGS BABYYYYY

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-05-27 16:01 UTC	Run
prod-us	✅ Deployed	2026-05-27 16:16 UTC	Run
prod-eu	✅ Deployed	2026-05-27 16:16 UTC	Run