Cascade Pattern: Dependency Propagation and Impact Analysis

“

Series: S10 Design Patterns · Article 8 | Level: Advanced | Reading Time: 18 min

#TL;DR

The Cascade Pattern defines how dependency changes propagate between Ontology objects — when one object changes, how should all objects that depend on it respond.
coomia-dip supports four cascade strategies: Cascade Update, Cascade Delete, Restrict, and Set Null, declared in LinkType definitions.
Through dependency DAG (Directed Acyclic Graph) and topological sorting, coomia-dip ensures cascade operations execute in the correct order and provides pre-change Impact Analysis capabilities.

#Introduction: The Butterfly Effect

In ontology-driven systems, objects establish rich relationships through LinkTypes. Modifying one object can trigger chain reactions:

Code

Scenario: Deleting a Department
→ What happens to Employees under that department?
→ What happens to Assets owned by those employees?
→ What happens to MaintenanceRecords linked to those assets?
→ What happens to Projects those employees participate in?
→ What happens to Budgets associated with those projects?

Without explicit cascade strategies, developers must either manually handle each dependency layer (error-prone) or ignore dependencies leading to data inconsistency (dangling references).

The Cascade Pattern declares these strategies in the Schema, with the platform executing them automatically.

#Part 1: Cascade Strategy Definitions

#1.1 Four Cascade Strategies

Python

from dataclasses import dataclass, field
from enum import Enum
from typing import Any

class CascadeStrategy(Enum):
    CASCADE = "cascade"       # Cascade: delete children when parent is deleted
    RESTRICT = "restrict"     # Restrict: reject operation if dependents exist
    SET_NULL = "set_null"     # Set null: set dependent references to null
    SET_DEFAULT = "set_default"  # Set default: set dependent references to default
    NO_ACTION = "no_action"   # No action: do nothing (dangerous)

@dataclass
class CascadeRule:
    """Cascade rule defined on a LinkType."""
    link_type: str
    source_type: str
    target_type: str
    on_delete: CascadeStrategy = CascadeStrategy.RESTRICT
    on_update: CascadeStrategy = CascadeStrategy.CASCADE
    on_archive: CascadeStrategy = CascadeStrategy.CASCADE
    depth_limit: int = 10
    async_execution: bool = False
    batch_size: int = 1000

@dataclass
class CascadeConfig:
    object_type: str
    rules: list[CascadeRule] = field(default_factory=list)

    def get_rules_for_source(self, source_type: str) -> list[CascadeRule]:
        return [r for r in self.rules if r.source_type == source_type]

#1.2 Declaring Cascade Strategies in LinkTypes

Python

department_employee_link = CascadeRule(
    link_type="Department_employs_Employee",
    source_type="Department",
    target_type="Employee",
    on_delete=CascadeStrategy.RESTRICT,
    on_update=CascadeStrategy.CASCADE,
    on_archive=CascadeStrategy.CASCADE,
)

employee_asset_link = CascadeRule(
    link_type="Employee_owns_Asset",
    source_type="Employee",
    target_type="Asset",
    on_delete=CascadeStrategy.SET_NULL,
    on_update=CascadeStrategy.CASCADE,
)

project_budget_link = CascadeRule(
    link_type="Project_has_Budget",
    source_type="Project",
    target_type="Budget",
    on_delete=CascadeStrategy.CASCADE,
    on_update=CascadeStrategy.CASCADE,
)

#Part 2: Dependency Graph and Topological Sorting

#2.1 Building the Dependency DAG

coomia-dip builds a global dependency DAG from LinkTypes and CascadeRules at startup:

Python

class DependencyDAG:
    """Directed Acyclic Graph for object dependencies."""

    def __init__(self):
        self._edges: dict[str, list[tuple[str, CascadeRule]]] = {}
        self._reverse_edges: dict[str, list[tuple[str, CascadeRule]]] = {}

    def add_dependency(self, rule: CascadeRule) -> None:
        self._edges.setdefault(rule.source_type, []).append(
            (rule.target_type, rule)
        )
        self._reverse_edges.setdefault(rule.target_type, []).append(
            (rule.source_type, rule)
        )

    def get_dependents(self, object_type: str) -> list[tuple[str, CascadeRule]]:
        return self._edges.get(object_type, [])

    def get_dependencies(self, object_type: str) -> list[tuple[str, CascadeRule]]:
        return self._reverse_edges.get(object_type, [])

    def topological_sort(self) -> list[str]:
        """Return types in topological order (parents before children)."""
        in_degree: dict[str, int] = {}
        all_types: set[str] = set()

        for source, targets in self._edges.items():
            all_types.add(source)
            for target, _ in targets:
                all_types.add(target)
                in_degree[target] = in_degree.get(target, 0) + 1

        queue = [t for t in all_types if in_degree.get(t, 0) == 0]
        result = []

        while queue:
            node = queue.pop(0)
            result.append(node)
            for target, _ in self._edges.get(node, []):
                in_degree[target] -= 1
                if in_degree[target] == 0:
                    queue.append(target)

        if len(result) != len(all_types):
            raise CyclicDependencyError(
                "Circular dependency detected in Ontology schema"
            )
        return result

    def get_cascade_order(self, object_type: str) -> list[str]:
        visited: set[str] = set()
        order: list[str] = []

        def dfs(current: str, depth: int = 0) -> None:
            if current in visited or depth > 20:
                return
            visited.add(current)
            for target, _ in self.get_dependents(current):
                dfs(target, depth + 1)
            order.append(current)

        dfs(object_type)
        return list(reversed(order))

#2.2 Circular Dependency Detection

coomia-dip automatically detects circular dependencies during Schema registration to prevent infinite cascades:

Python

class CyclicDependencyDetector:
    """Detect circular dependencies in the Ontology schema."""

    def detect(self, dag: DependencyDAG) -> list[list[str]]:
        cycles: list[list[str]] = []
        visited: set[str] = set()
        rec_stack: set[str] = set()

        def dfs(node: str, path: list[str]) -> None:
            visited.add(node)
            rec_stack.add(node)
            path.append(node)

            for target, _ in dag.get_dependents(node):
                if target not in visited:
                    dfs(target, path)
                elif target in rec_stack:
                    cycle_start = path.index(target)
                    cycles.append(path[cycle_start:] + [target])

            path.pop()
            rec_stack.discard(node)

        all_types = set()
        for source in dag._edges:
            all_types.add(source)
            for target, _ in dag._edges[source]:
                all_types.add(target)

        for node in all_types:
            if node not in visited:
                dfs(node, [])
        return cycles

#Part 3: Cascade Execution Engine

#3.1 Cascade Delete

Python

class CascadeEngine:
    """Engine for executing cascade operations."""

    def __init__(self, dag: DependencyDAG, repository: "ObjectRepository", event_store: "EventStore"):
        self._dag = dag
        self._repo = repository
        self._event_store = event_store

    async def cascade_delete(
        self, object_type: str, object_id: str, context: "OperationContext", dry_run: bool = False
    ) -> "CascadeResult":
        # Phase 1: Impact analysis
        impact = await self._analyze_impact(object_type, object_id, "delete")

        if dry_run:
            return CascadeResult(status="dry_run", impact=impact, executed=False)

        # Phase 2: Check RESTRICT strategies
        for dep_type, rule in self._dag.get_dependents(object_type):
            if rule.on_delete == CascadeStrategy.RESTRICT:
                count = await self._repo.count_by_reference(dep_type, object_type, object_id)
                if count > 0:
                    return CascadeResult(
                        status="restricted", impact=impact, executed=False,
                        error=f"Cannot delete: {count} {dep_type} objects depend on this",
                    )

        # Phase 3: Execute cascade in reverse topological order (children first)
        cascade_order = self._dag.get_cascade_order(object_type)
        deleted_objects: list[dict] = []

        for dep_type in reversed(cascade_order):
            if dep_type == object_type:
                continue
            rule = self._get_rule(object_type, dep_type)
            if not rule:
                continue

            dependents = await self._repo.find_by_reference(dep_type, object_type, object_id)

            for dep_obj in dependents:
                if rule.on_delete == CascadeStrategy.CASCADE:
                    await self.cascade_delete(dep_type, dep_obj["_id"], context)
                    deleted_objects.append(dep_obj)
                elif rule.on_delete == CascadeStrategy.SET_NULL:
                    await self._repo.set_reference_null(dep_type, dep_obj["_id"], object_type)
                elif rule.on_delete == CascadeStrategy.SET_DEFAULT:
                    default = rule.metadata.get("default_value")
                    await self._repo.set_reference(dep_type, dep_obj["_id"], object_type, default)

        # Phase 4: Delete the target object
        await self._repo.delete(object_type, object_id)

        # Phase 5: Record event
        await self._event_store.append([
            DomainEvent(
                event_id=generate_id(),
                event_type="cascade.delete",
                aggregate_id=object_id,
                aggregate_type=object_type,
                sequence_number=0,
                timestamp=datetime.utcnow(),
                payload={"cascade_deleted": len(deleted_objects), "impact": impact},
                metadata=EventMetadata(
                    actor_id=context.actor_id, actor_type=context.actor_type,
                    tenant_id=context.tenant_id, world_id=context.world_id,
                    source_plane="control", trace_id=context.trace_id,
                ),
            )
        ])

        return CascadeResult(
            status="completed", impact=impact, executed=True,
            deleted_count=len(deleted_objects) + 1,
        )

#3.2 Cascade Update

Python

class CascadeUpdateEngine:
    """Handle cascade updates when object properties change."""

    async def cascade_update(
        self, object_type: str, object_id: str, changes: dict[str, Any], context: "OperationContext"
    ) -> "CascadeResult":
        cascadable_changes = self._filter_cascadable(object_type, changes)
        if not cascadable_changes:
            return CascadeResult(status="no_cascade_needed", executed=False)

        updated_objects: list[dict] = []

        for dep_type, rule in self._dag.get_dependents(object_type):
            if rule.on_update != CascadeStrategy.CASCADE:
                continue

            dependents = await self._repo.find_by_reference(dep_type, object_type, object_id)

            for dep_obj in dependents:
                dep_changes = self._compute_dependent_changes(rule, changes, dep_obj)
                if dep_changes:
                    await self._repo.update(dep_type, dep_obj["_id"], dep_changes)
                    updated_objects.append({
                        "type": dep_type, "id": dep_obj["_id"], "changes": dep_changes,
                    })
                    await self.cascade_update(dep_type, dep_obj["_id"], dep_changes, context)

        return CascadeResult(
            status="completed", executed=True,
            updated_count=len(updated_objects),
            impact={"updated_objects": updated_objects},
        )

#Part 4: Impact Analysis

#4.1 Pre-Change Impact Assessment

Before executing cascade operations, coomia-dip provides impact analysis to help users understand how many objects will be affected:

Python

class ImpactAnalyzer:
    """Analyze the impact of cascade operations before execution."""

    async def analyze(self, object_type: str, object_id: str, operation: str) -> dict:
        impact: dict[str, Any] = {
            "root": {"type": object_type, "id": object_id},
            "operation": operation,
            "affected_types": {},
            "total_affected": 0,
            "risk_level": "low",
        }

        await self._collect_impact(object_type, object_id, operation, impact, depth=0)

        if impact["total_affected"] > 1000:
            impact["risk_level"] = "critical"
        elif impact["total_affected"] > 100:
            impact["risk_level"] = "high"
        elif impact["total_affected"] > 10:
            impact["risk_level"] = "medium"

        return impact

    async def _collect_impact(
        self, object_type: str, object_id: str, operation: str, impact: dict, depth: int
    ) -> None:
        if depth > 10:
            return

        for dep_type, rule in self._dag.get_dependents(object_type):
            strategy = getattr(rule, f"on_{operation}", CascadeStrategy.NO_ACTION)
            count = await self._repo.count_by_reference(dep_type, object_type, object_id)

            if count > 0:
                impact["affected_types"][dep_type] = {
                    "count": count, "strategy": strategy.value, "depth": depth + 1,
                }
                impact["total_affected"] += count

                if strategy == CascadeStrategy.CASCADE:
                    dependents = await self._repo.find_by_reference(
                        dep_type, object_type, object_id
                    )
                    for dep in dependents[:10]:
                        await self._collect_impact(
                            dep_type, dep["_id"], operation, impact, depth + 1
                        )

#4.2 Impact Analysis API in SDK

Python

from ontology_sdk import OntoPlatform

client = OntoPlatform.connect("https://platform.example.com")

# Analyze impact before deletion
impact = client.objects.analyze_delete_impact(
    object_type="Department",
    object_id="dept-001",
)

print(f"Risk level: {impact.risk_level}")
print(f"Total affected: {impact.total_affected}")

for type_name, info in impact.affected_types.items():
    print(f"  {type_name}: {info.count} objects ({info.strategy})")

# Execute deletion after confirmation
if impact.risk_level in ("low", "medium"):
    result = client.objects.delete(
        object_type="Department",
        object_id="dept-001",
        cascade=True,
    )

#Part 5: Async Execution for Large-Scale Cascades

#5.1 Async Cascade Tasks

When cascade operations affect a large number of objects, synchronous execution may cause timeouts. coomia-dip supports async cascades:

Python

class AsyncCascadeExecutor:
    async def submit(
        self, object_type: str, object_id: str, operation: str, context: "OperationContext"
    ) -> str:
        task_id = generate_id()
        await self._temporal.start_workflow(
            CascadeWorkflow,
            args={
                "task_id": task_id, "object_type": object_type,
                "object_id": object_id, "operation": operation, "context": context,
            },
            id=f"cascade-{task_id}",
        )
        return task_id

    async def get_progress(self, task_id: str) -> dict:
        return await self._progress_store.get(task_id)

#5.2 Batch Cascade Processing

Python

class BatchCascadeProcessor:
    async def process_batch(
        self, dep_type: str, dep_objects: list[dict], rule: CascadeRule,
        operation: str, batch_size: int = 1000
    ) -> dict:
        total = len(dep_objects)
        processed = 0
        errors = []

        for i in range(0, total, batch_size):
            batch = dep_objects[i:i + batch_size]
            for obj in batch:
                try:
                    if operation == "delete" and rule.on_delete == CascadeStrategy.CASCADE:
                        await self._repo.delete(dep_type, obj["_id"])
                    elif operation == "delete" and rule.on_delete == CascadeStrategy.SET_NULL:
                        await self._repo.set_reference_null(dep_type, obj["_id"], rule.source_type)
                    processed += 1
                except Exception as e:
                    errors.append({"id": obj["_id"], "error": str(e)})

            await self._progress_reporter.report(
                processed=processed, total=total, errors=len(errors)
            )

        return {"total": total, "processed": processed, "errors": errors}

#Part 6: Derived Property Cascade Recalculation

#6.1 Derived Property Dependencies

When source properties change, derived properties that depend on them need recalculation. This is a special form of cascade update:

Python

class DerivedPropertyCascade:
    async def on_property_change(
        self, object_type: str, object_id: str, changed_property: str, new_value: Any
    ) -> list[dict]:
        affected = self._dependency_graph.get_derived_from(object_type, changed_property)
        recalculated = []

        for derived in affected:
            new_derived_value = await self._calculator.compute(
                derived.formula, object_type=object_type, object_id=object_id,
            )
            await self._repo.update(
                derived.target_type, object_id,
                {derived.property_name: new_derived_value},
            )
            recalculated.append({
                "property": derived.property_name,
                "old_value": derived.current_value,
                "new_value": new_derived_value,
            })
            # Recursive: this derived property change may trigger more cascades
            await self.on_property_change(
                derived.target_type, object_id,
                derived.property_name, new_derived_value,
            )

        return recalculated

#Key Takeaways

Schema-Declarative: Cascade strategies are declared in LinkType Schemas and executed automatically by the platform
Four Strategies: CASCADE, RESTRICT, SET_NULL, and SET_DEFAULT cover all common scenarios
Topological Sorting: Dependency DAG and topological sorting ensure cascade operations execute in the correct order
Impact Analysis: Pre-change impact assessment lets users understand the scope and risk level of operations
Async Execution: Large-scale cascades use async workflows and batch processing to avoid timeouts
Derived Properties: Source property changes automatically trigger derived property recalculation, forming complete cascade chains

#Next Article

In the next article, we will explore the Query Rewrite pattern — how coomia-dip translates high-level semantic queries into optimized storage engine queries.

S10-09: Query Rewrite: Optimized Translation from Semantics to Storage

#Tags

#DesignPatterns #CascadePattern #DependencyPropagation #ImpactAnalysis #TopologicalSort #DAG #DerivedProperties #CascadeDelete