feat: store raw processor tree JSON and add error categorization fields

Fixes iteration overlay corruption caused by flat storage collapsing duplicate processorIds across loop iterations. Server: - Store raw processor tree as processors_json JSONB on executions table - Detail endpoint serves from processors_json (faithful tree), falls back to flat record reconstruction for older executions - V10 migration: processors_json, error categorization (errorType, errorCategory, rootCauseType, rootCauseMessage), OTel (traceId, spanId), circuit breaker (circuitBreakerState, fallbackTriggered), drops erroneous splitDepth/loopDepth columns - Add all new fields through full ingestion/storage/API chain UI: - Fix overlay wrapper filtering: check wrapper type before status filter - Add new fields to schema.d.ts Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-28 21:44:54 +01:00
parent f12f5f3c8d
commit 30344d29b1
12 changed files with 226 additions and 48 deletions
--- a/cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/PostgresExecutionStore.java
+++ b/cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/PostgresExecutionStore.java
@@ -28,8 +28,13 @@ public class PostgresExecutionStore implements ExecutionStore {
                    status, correlation_id, exchange_id, start_time, end_time,
                    duration_ms, error_message, error_stacktrace, diagram_content_hash,
                    engine_level, input_body, output_body, input_headers, output_headers,
-                    attributes, created_at, updated_at)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?::jsonb, now(), now())
+                    attributes,
+                    error_type, error_category, root_cause_type, root_cause_message,
+                    trace_id, span_id,
+                    processors_json,
+                    created_at, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?::jsonb,
+                        ?, ?, ?, ?, ?, ?, ?::jsonb, now(), now())
                ON CONFLICT (execution_id, start_time) DO UPDATE SET
                    status = CASE
                        WHEN EXCLUDED.status IN ('COMPLETED', 'FAILED')
@@ -49,6 +54,13 @@ public class PostgresExecutionStore implements ExecutionStore {
                    input_headers = COALESCE(EXCLUDED.input_headers, executions.input_headers),
                    output_headers = COALESCE(EXCLUDED.output_headers, executions.output_headers),
                    attributes = COALESCE(EXCLUDED.attributes, executions.attributes),
+                    error_type = COALESCE(EXCLUDED.error_type, executions.error_type),
+                    error_category = COALESCE(EXCLUDED.error_category, executions.error_category),
+                    root_cause_type = COALESCE(EXCLUDED.root_cause_type, executions.root_cause_type),
+                    root_cause_message = COALESCE(EXCLUDED.root_cause_message, executions.root_cause_message),
+                    trace_id = COALESCE(EXCLUDED.trace_id, executions.trace_id),
+                    span_id = COALESCE(EXCLUDED.span_id, executions.span_id),
+                    processors_json = COALESCE(EXCLUDED.processors_json, executions.processors_json),
                    updated_at = now()
                """,
                execution.executionId(), execution.routeId(), execution.agentId(),
@@ -61,7 +73,11 @@ public class PostgresExecutionStore implements ExecutionStore {
                execution.engineLevel(),
                execution.inputBody(), execution.outputBody(),
                execution.inputHeaders(), execution.outputHeaders(),
-                execution.attributes());
+                execution.attributes(),
+                execution.errorType(), execution.errorCategory(),
+                execution.rootCauseType(), execution.rootCauseMessage(),
+                execution.traceId(), execution.spanId(),
+                execution.processorsJson());
    }

    @Override
@@ -74,8 +90,11 @@ public class PostgresExecutionStore implements ExecutionStore {
                    status, start_time, end_time, duration_ms, error_message, error_stacktrace,
                    input_body, output_body, input_headers, output_headers, attributes,
                    loop_index, loop_size, split_index, split_size, multicast_index,
-                    resolved_endpoint_uri, split_depth, loop_depth)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?::jsonb, ?, ?, ?, ?, ?, ?, ?, ?)
+                    resolved_endpoint_uri,
+                    error_type, error_category, root_cause_type, root_cause_message,
+                    error_handler_type, circuit_breaker_state, fallback_triggered)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?::jsonb,
+                        ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT (execution_id, processor_id, start_time) DO UPDATE SET
                    status = EXCLUDED.status,
                    end_time = COALESCE(EXCLUDED.end_time, processor_executions.end_time),
@@ -93,8 +112,13 @@ public class PostgresExecutionStore implements ExecutionStore {
                    split_size = COALESCE(EXCLUDED.split_size, processor_executions.split_size),
                    multicast_index = COALESCE(EXCLUDED.multicast_index, processor_executions.multicast_index),
                    resolved_endpoint_uri = COALESCE(EXCLUDED.resolved_endpoint_uri, processor_executions.resolved_endpoint_uri),
-                    split_depth = EXCLUDED.split_depth,
-                    loop_depth = EXCLUDED.loop_depth
+                    error_type = COALESCE(EXCLUDED.error_type, processor_executions.error_type),
+                    error_category = COALESCE(EXCLUDED.error_category, processor_executions.error_category),
+                    root_cause_type = COALESCE(EXCLUDED.root_cause_type, processor_executions.root_cause_type),
+                    root_cause_message = COALESCE(EXCLUDED.root_cause_message, processor_executions.root_cause_message),
+                    error_handler_type = COALESCE(EXCLUDED.error_handler_type, processor_executions.error_handler_type),
+                    circuit_breaker_state = COALESCE(EXCLUDED.circuit_breaker_state, processor_executions.circuit_breaker_state),
+                    fallback_triggered = COALESCE(EXCLUDED.fallback_triggered, processor_executions.fallback_triggered)
                """,
                processors.stream().map(p -> new Object[]{
                        p.executionId(), p.processorId(), p.processorType(),
@@ -108,8 +132,10 @@ public class PostgresExecutionStore implements ExecutionStore {
                        p.loopIndex(), p.loopSize(), p.splitIndex(), p.splitSize(),
                        p.multicastIndex(),
                        p.resolvedEndpointUri(),
-                        p.splitDepth(),
-                        p.loopDepth()
+                        p.errorType(), p.errorCategory(),
+                        p.rootCauseType(), p.rootCauseMessage(),
+                        p.errorHandlerType(), p.circuitBreakerState(),
+                        p.fallbackTriggered()
                }).toList());
    }

@@ -148,7 +174,11 @@ public class PostgresExecutionStore implements ExecutionStore {
                    rs.getString("engine_level"),
                    rs.getString("input_body"), rs.getString("output_body"),
                    rs.getString("input_headers"), rs.getString("output_headers"),
-                    rs.getString("attributes"));
+                    rs.getString("attributes"),
+                    rs.getString("error_type"), rs.getString("error_category"),
+                    rs.getString("root_cause_type"), rs.getString("root_cause_message"),
+                    rs.getString("trace_id"), rs.getString("span_id"),
+                    rs.getString("processors_json"));

    private static final RowMapper<ProcessorRecord> PROCESSOR_MAPPER = (rs, rowNum) ->
            new ProcessorRecord(
@@ -169,8 +199,10 @@ public class PostgresExecutionStore implements ExecutionStore {
                    rs.getObject("split_size") != null ? rs.getInt("split_size") : null,
                    rs.getObject("multicast_index") != null ? rs.getInt("multicast_index") : null,
                    rs.getString("resolved_endpoint_uri"),
-                    rs.getInt("split_depth"),
-                    rs.getInt("loop_depth"));
+                    rs.getString("error_type"), rs.getString("error_category"),
+                    rs.getString("root_cause_type"), rs.getString("root_cause_message"),
+                    rs.getString("error_handler_type"), rs.getString("circuit_breaker_state"),
+                    rs.getObject("fallback_triggered") != null ? rs.getBoolean("fallback_triggered") : null);

    private static Instant toInstant(ResultSet rs, String column) throws SQLException {
        Timestamp ts = rs.getTimestamp(column);
--- a/cameleer3-server-app/src/main/resources/db/migration/V10__error_categorization_and_tracing.sql
+++ b/cameleer3-server-app/src/main/resources/db/migration/V10__error_categorization_and_tracing.sql
@@ -0,0 +1,23 @@
+-- executions: store raw processor tree for faithful detail response
+ALTER TABLE executions ADD COLUMN processors_json JSONB;
+
+-- executions: error categorization + OTel tracing
+ALTER TABLE executions ADD COLUMN error_type TEXT;
+ALTER TABLE executions ADD COLUMN error_category TEXT;
+ALTER TABLE executions ADD COLUMN root_cause_type TEXT;
+ALTER TABLE executions ADD COLUMN root_cause_message TEXT;
+ALTER TABLE executions ADD COLUMN trace_id TEXT;
+ALTER TABLE executions ADD COLUMN span_id TEXT;
+
+-- processor_executions: error categorization + circuit breaker
+ALTER TABLE processor_executions ADD COLUMN error_type TEXT;
+ALTER TABLE processor_executions ADD COLUMN error_category TEXT;
+ALTER TABLE processor_executions ADD COLUMN root_cause_type TEXT;
+ALTER TABLE processor_executions ADD COLUMN root_cause_message TEXT;
+ALTER TABLE processor_executions ADD COLUMN error_handler_type TEXT;
+ALTER TABLE processor_executions ADD COLUMN circuit_breaker_state TEXT;
+ALTER TABLE processor_executions ADD COLUMN fallback_triggered BOOLEAN;
+
+-- Remove erroneous depth columns from V9
+ALTER TABLE processor_executions DROP COLUMN IF EXISTS split_depth;
+ALTER TABLE processor_executions DROP COLUMN IF EXISTS loop_depth;
--- a/cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/PostgresExecutionStoreIT.java
+++ b/cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/PostgresExecutionStoreIT.java
@@ -26,7 +26,8 @@ class PostgresExecutionStoreIT extends AbstractPostgresIT {
                "COMPLETED", "corr-1", "exchange-1",
                now, now.plusMillis(100), 100L,
                null, null, null,
-                "REGULAR", null, null, null, null, null);
+                "REGULAR", null, null, null, null, null,
+                null, null, null, null, null, null, null);

        executionStore.upsert(record);
        Optional<ExecutionRecord> found = executionStore.findById("exec-1");
@@ -43,11 +44,13 @@ class PostgresExecutionStoreIT extends AbstractPostgresIT {
        ExecutionRecord first = new ExecutionRecord(
                "exec-dup", "route-a", "agent-1", "app-1",
                "RUNNING", null, null, now, null, null, null, null, null,
-                null, null, null, null, null, null);
+                null, null, null, null, null, null,
+                null, null, null, null, null, null, null);
        ExecutionRecord second = new ExecutionRecord(
                "exec-dup", "route-a", "agent-1", "app-1",
                "COMPLETED", null, null, now, now.plusMillis(200), 200L, null, null, null,
-                "COMPLETE", null, null, null, null, null);
+                "COMPLETE", null, null, null, null, null,
+                null, null, null, null, null, null, null);

        executionStore.upsert(first);
        executionStore.upsert(second);
@@ -64,7 +67,8 @@ class PostgresExecutionStoreIT extends AbstractPostgresIT {
        ExecutionRecord exec = new ExecutionRecord(
                "exec-proc", "route-a", "agent-1", "app-1",
                "COMPLETED", null, null, now, now.plusMillis(50), 50L, null, null, null,
-                "COMPLETE", null, null, null, null, null);
+                "COMPLETE", null, null, null, null, null,
+                null, null, null, null, null, null, null);
        executionStore.upsert(exec);

        List<ProcessorRecord> processors = List.of(
@@ -73,13 +77,13 @@ class PostgresExecutionStoreIT extends AbstractPostgresIT {
                        now, now.plusMillis(10), 10L, null, null,
                        "input body", "output body", null, null, null,
                        null, null, null, null, null,
-                        null, 0, 0),
+                        null, null, null, null, null, null, null, null),
                new ProcessorRecord("exec-proc", "proc-2", "to",
                        "app-1", "route-a", 1, "proc-1", "COMPLETED",
                        now.plusMillis(10), now.plusMillis(30), 20L, null, null,
                        null, null, null, null, null,
                        null, null, null, null, null,
-                        null, 0, 0)
+                        null, null, null, null, null, null, null, null)
        );
        executionStore.upsertProcessors("exec-proc", now, "app-1", "route-a", processors);

--- a/cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/PostgresStatsStoreIT.java
+++ b/cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/PostgresStatsStoreIT.java
@@ -60,6 +60,7 @@ class PostgresStatsStoreIT extends AbstractPostgresIT {
                id, routeId, "agent-1", applicationName, status, null, null,
                startTime, startTime.plusMillis(durationMs), durationMs,
                status.equals("FAILED") ? "error" : null, null, null,
-                null, null, null, null, null, null));
+                null, null, null, null, null, null,
+                null, null, null, null, null, null, null));
    }
 }