Downstream Sync Triggers Not Working - Investigation Report

Date: December 7, 2025
Status: 🔴 Critical Issue - Sensors Not Running

Executive Summary

Bronze syncs are completing successfully, but downstream syncs (dbt transform → silver/gold, and Typesense sync) are not being triggered. Investigation reveals that the event-driven sensors (bronze_to_dbt_sensor and dbt_to_typesense_sensor) are not running at all - they don’t appear in Dagster daemon logs.

Problem Statement

Expected Behavior:

1. Bronze assets materialize → bronze_to_dbt_sensor triggers → dbt transform runs
2. Gold assets materialize → dbt_to_typesense_sensor triggers → Typesense sync runs

Actual Behavior:

• Bronze syncs complete successfully (confirmed in logs)
• No sensor evaluation logs for bronze_to_dbt_sensor or dbt_to_typesense_sensor
• No dbt transform runs triggered
• No Typesense sync runs triggered

Evidence

1. Bronze Materializations Are Happening

From logs (12:00-12:27 UTC):

2025-12-07 12:00:46 - bronze__customers - ASSET_MATERIALIZATION
2025-12-07 12:00:48 - bronze__products - ASSET_MATERIALIZATION
2025-12-07 12:00:48 - bronze__customers - ASSET_MATERIALIZATION
2025-12-07 12:00:49 - bronze__cross_references - ASSET_MATERIALIZATION
2025-12-07 12:00:50 - bronze__ship_to_links - ASSET_MATERIALIZATION
... (many more)
2025-12-07 12:25:46 - bronze__products - ASSET_MATERIALIZATION
2025-12-07 12:27:08 - bronze__products - ASSET_MATERIALIZATION

Status: ✅ Bronze syncs are working correctly

2. Sensors Are Registered But Not Running

Registered Sensors (from definitions.py):

• ✅ erp_connection_discovery_sensor - RUNNING (every 5 minutes)
• ✅ sync_failure_alert - RUNNING (every 30 seconds)
• ❌ bronze_to_dbt_sensor - NOT RUNNING (no logs found)
• ❌ dbt_to_typesense_sensor - NOT RUNNING (no logs found)

Log Evidence:

2025-12-07 12:15:12 - Checking for new runs for sensor: erp_connection_discovery_sensor
2025-12-07 12:15:17 - Checking for new runs for sensor: sync_failure_alert
2025-12-07 12:20:12 - Checking for new runs for sensor: erp_connection_discovery_sensor
2025-12-07 12:20:17 - Checking for new runs for sensor: sync_failure_alert
... (repeating pattern)

Missing: No logs for bronze_to_dbt_sensor or dbt_to_typesense_sensor evaluation

3. Sensor Configuration

Code Location: apps/dagster/erp_pipeline/sensors.py

bronze_to_dbt_sensor:

• Name: bronze_to_dbt_sensor
• Minimum interval: 30 seconds
• Job: dbt_transform_job
• Watches: All bronze asset keys (products, customers, cross_references, shipping_addresses, ship_to_links)
• Trigger: On ASSET_MATERIALIZATION events

dbt_to_typesense_sensor:

• Name: dbt_to_typesense_sensor
• Minimum interval: 30 seconds
• Job: typesense_sync_job
• Watches: All gold asset keys (dim_products, dim_customers, xref_products, dim_shipping_addresses)
• Trigger: On ASSET_MATERIALIZATION events

Registration: ✅ Sensors are exported in all_event_sensors and included in definitions.py

Root Cause Analysis

Hypothesis 1: Sensors Not Enabled in Dagster UI

Likelihood: HIGH Evidence: Sensors are registered in code but may not be enabled in Dagster instance

How to Check:

1. Go to Dagster UI → Automation → Sensors
2. Check if bronze_to_dbt_sensor and dbt_to_typesense_sensor are listed
3. Check their status (should be “Running” or “Stopped”)

Fix: Enable sensors in Dagster UI if they’re stopped

Hypothesis 2: Sensor Evaluation Errors (Silent Failures)

Likelihood: MEDIUM Evidence: No error logs found, but sensors might be failing during evaluation

Potential Issues:

1. Event Query API Issue: The sensor uses get_event_records() with after_cursor - this might not work correctly with integer cursors
2. Partition Tag Issue: Sensor looks for "dagster/partition" in run tags - this might not be present for partitioned runs
3. Asset Key Mismatch: The asset keys in BRONZE_ASSET_KEYS might not match the actual asset keys

Code Concern:

# Line 95 in sensors.py
if run and "dagster/partition" in run.tags:
    partition_key = run.tags["dagster/partition"]

The partition tag might be stored differently. Should check:

• run.tags.get("dagster/partition")
• Or use context.partition_key from the materialization event itself

Hypothesis 3: Sensor Daemon Not Picking Up New Sensors

Likelihood: LOW Evidence: Other sensors are running, so daemon is working

Possible Causes:

• Sensors added after daemon started (requires restart)
• Sensor definition syntax error preventing registration
• Sensor name conflict

Hypothesis 4: Event Filter Not Matching

Likelihood: MEDIUM Evidence: Bronze materializations exist but sensor isn’t finding them

Potential Issues:

1. Cursor Issue: Starting cursor is "0" which might not match actual event IDs
2. Event Type: DagsterEventType.ASSET_MATERIALIZATION might not be the correct type
3. Asset Key Format: Asset keys might need different format (e.g., ["bronze", "products"] vs "bronze__products")

Code Analysis

Sensor Implementation Issues

Issue 1: Partition Key Extraction

# Current code (line 94-96)
run = context.instance.get_run_by_id(record.run_id)
if run and "dagster/partition" in run.tags:
    partition_key = run.tags["dagster/partition"]

Problem: The partition key might be in a different tag format or in the event metadata itself.

Better Approach:

• Check event metadata for partition information
• Use record.dagster_event.asset_materialization.partition if available
• Or query the asset materialization metadata directly

Issue 2: Cursor Management

cursor = context.cursor or "0"
last_event_id = int(cursor)

Problem: Converting cursor to int might fail if cursor format is different, or if cursor is None.

Better Approach:

• Add error handling for cursor conversion
• Log cursor value for debugging
• Handle None cursor more gracefully

Issue 3: Event Query Filter

records = context.instance.get_event_records(
    EventRecordsFilter(
        event_type=DagsterEventType.ASSET_MATERIALIZATION,
        asset_key=asset_key,
        after_cursor=last_event_id,
    ),
    limit=100,
)

Problem: after_cursor expects a string, not an integer. This might cause the query to fail silently.

Fix: Use after_cursor=str(last_event_id) or check Dagster API documentation for correct format.

Recommended Investigation Steps

Step 1: Check Sensor Status in Dagster UI

1. Navigate to Dagster UI → Automation → Sensors
2. Verify bronze_to_dbt_sensor and dbt_to_typesense_sensor are listed
3. Check if they’re enabled/running
4. If stopped, enable them

Step 2: Add Debug Logging to Sensors

Add logging at the start of sensor evaluation:

def bronze_to_dbt_sensor(context: SensorEvaluationContext) -> SensorResult:
    context.log.info("bronze_to_dbt_sensor: Starting evaluation")
    # ... rest of code

This will confirm if sensors are being evaluated at all.

Step 3: Check Event Query Results

Add logging to see what events are found:

for asset_key in BRONZE_ASSET_KEYS:
    records = context.instance.get_event_records(...)
    context.log.info(f"Found {len(records)} records for {asset_key}")
    # ... rest of code

Step 4: Verify Asset Key Format

Check if asset keys match actual materialized assets:

• Log materialized asset keys from bronze runs
• Compare with BRONZE_ASSET_KEYS in sensor code
• Ensure format matches exactly

Step 5: Test Sensor Manually

1. Manually trigger a bronze sync
2. Immediately check sensor logs
3. Verify sensor evaluation happens within 30 seconds
4. Check if events are found

Immediate Action Items

1. Check Dagster UI - Verify sensor status and enable if needed
2. Add Debug Logging - Add logging to sensor functions to confirm evaluation
3. Fix Cursor Format - Ensure after_cursor uses correct format (string vs int)
4. Fix Partition Extraction - Use correct method to get partition from materialization events
5. Test Manually - Trigger bronze sync and monitor sensor logs

Files to Review

• apps/dagster/erp_pipeline/sensors.py - Sensor implementations
• apps/dagster/erp_pipeline/definitions.py - Sensor registration
• Dagster UI → Automation → Sensors - Sensor status

Expected Fix

Once sensors are enabled and working:

1. Bronze materialization → Sensor detects event → Logs “Bronze materialization for partition X”
2. Sensor creates RunRequest → Logs “Triggering dbt transform for N partitions”
3. dbt_transform_job runs → Gold assets materialize
4. dbt_to_typesense_sensor detects gold materialization → Triggers Typesense sync

---

Report Generated: 2025-12-07 12:35 UTC
Investigator: AI Assistant
Status: Ready for Engineering Team Review