Postmortem: How a Lattice 2.9 Bug Caused Our Promotion Cycle to Be Delayed by 2 Weeks

import logging
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set
from lattice_sdk import LatticeClient, RBACGroup, PromotionWorkflow, WorkflowStatus
from lattice_sdk.exceptions import LatticeAPIError, CacheInvalidationError

# Configure logging to capture debug-level RBAC cache events
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

@dataclass
class PromotionApprovalEvent:
    """Represents a promotion approval event in Lattice 2.9"""
    event_id: str
    user_id: str
    role_group: str
    approver_ids: List[str]
    timestamp: float = field(default_factory=time.time)
    status: WorkflowStatus = WorkflowStatus.PENDING

def reproduce_rbac_cache_bug() -> None:
    """
    Reproduces the Lattice 2.9 RBAC cache invalidation bug that drops promotion approval events
    for nested role groups. This matches the exact workflow that failed in production on Oct 12.
    """
    try:
        # Initialize Lattice 2.9 client with production-matching config
        client = LatticeClient(
            api_key="prod-simulation-key",
            version="2.9.0",
            cache_ttl=300,  # 5 minute cache TTL matching production config
            enable_rbac_cache=True
        )
        logger.info("Initialized Lattice 2.9 client for bug reproduction")

        # Create nested role groups matching our production setup:
        # Engineering -> Backend -> Senior -> Staff (nested 4 levels deep)
        top_level_group = RBACGroup(id="eng-all", name="Engineering", parent=None)
        mid_level_group = RBACGroup(id="eng-backend", name="Backend Engineering", parent=top_level_group)
        senior_group = RBACGroup(id="eng-backend-senior", name="Senior Backend", parent=mid_level_group)
        staff_group = RBACGroup(id="eng-backend-staff", name="Staff Backend", parent=senior_group)

        # Register groups with Lattice (idempotent operation)
        for group in [top_level_group, mid_level_group, senior_group, staff_group]:
            try:
                client.rbac.create_group(group)
                logger.debug(f"Registered RBAC group: {group.id}")
            except LatticeAPIError as e:
                if e.code == 409:
                    logger.debug(f"Group {group.id} already exists, skipping")
                else:
                    raise

        # Create a promotion workflow for a staff engineer candidate
        workflow = PromotionWorkflow(
            workflow_id="promo-2024-1047",
            candidate_id="user-8923",
            target_role="Staff Backend Engineer",
            approval_group=staff_group.id,  # Nested 4 levels deep
            required_approvers=2
        )
        client.workflows.create(workflow)
        logger.info(f"Created promotion workflow: {workflow.workflow_id}")

        # Simulate 3 approval events (matches production load)
        approval_events: List[PromotionApprovalEvent] = []
        for i in range(3):
            event = PromotionApprovalEvent(
                event_id=f"evt-{workflow.workflow_id}-{i}",
                user_id=f"approver-{i}",
                role_group=staff_group.id,
                approver_ids=[f"approver-{i}"]
            )
            approval_events.append(event)
            try:
                # Submit approval event to Lattice 2.9 client
                client.workflows.submit_approval(event)
                logger.info(f"Submitted approval event {event.event_id}")
            except CacheInvalidationError as e:
                logger.error(f"Cache invalidation failed for event {event.event_id}: {e}")
                # This is the exact error that caused the production outage
                raise

        # Wait for cache to invalidate (production had 500ms retry window)
        time.sleep(0.5)

        # Check workflow status - in Lattice 2.9, this returns PENDING even with approvals
        status = client.workflows.get_status(workflow.workflow_id)
        if status == WorkflowStatus.PENDING:
            logger.error(
                f"BUG REPRODUCED: Workflow {workflow.workflow_id} still pending "
                f"after {len(approval_events)} approvals. Expected APPROVED."
            )
        elif status == WorkflowStatus.APPROVED:
            logger.info(f"Workflow {workflow.workflow_id} approved successfully")
        else:
            logger.warning(f"Unexpected workflow status: {status}")

    except LatticeAPIError as e:
        logger.error(f"Lattice API error: {e.code} - {e.message}")
        raise
    except Exception as e:
        logger.critical(f"Unexpected error during bug reproduction: {e}")
        raise

if __name__ == "__main__":
    reproduce_rbac_cache_bug()

import hashlib
import logging
import time
from typing import Dict, List, Optional, Set
from lattice_sdk.cache import BaseCache, CacheEntry
from lattice_sdk.rbac import RBACGroup, Permission

logger = logging.getLogger(__name__)

class PatchedRBACCache(BaseCache):
    """
    Patched RBAC cache implementation for Lattice >=2.9.3 that fixes nested group
    invalidation logic. This replaces the broken Lattice 2.9.0-2.9.2 cache implementation.
    """

    def __init__(self, ttl: int = 300, max_entries: int = 10000):
        super().__init__(ttl=ttl, max_entries=max_entries)
        self._nested_group_index: Dict[str, Set[str]] = {}  # group_id -> set of parent group IDs
        self._permission_index: Dict[str, Set[str]] = {}  # permission_id -> set of group IDs
        logger.info(f"Initialized PatchedRBACCache with TTL={ttl}s, max_entries={max_entries}")

    def _compute_cache_key(self, group_id: str, permission_id: Optional[str] = None) -> str:
        """Generate a deterministic cache key for a group/permission pair"""
        key_parts = [group_id]
        if permission_id:
            key_parts.append(permission_id)
        return hashlib.sha256("-".join(key_parts).encode()).hexdigest()

    def add_group(self, group: RBACGroup) -> None:
        """Add a group to the cache and update nested group index"""
        try:
            cache_key = self._compute_cache_key(group.id)
            entry = CacheEntry(
                key=cache_key,
                value=group,
                expires_at=time.time() + self.ttl
            )
            self._store[cache_key] = entry
            logger.debug(f"Added group {group.id} to cache with key {cache_key}")

            # Update nested group index for parent traversal
            if group.parent_id:
                if group.parent_id not in self._nested_group_index:
                    self._nested_group_index[group.parent_id] = set()
                self._nested_group_index[group.parent_id].add(group.id)
                logger.debug(f"Updated nested index: parent {group.parent_id} -> child {group.id}")

            # Index all permissions for the group
            for perm in group.permissions:
                if perm.id not in self._permission_index:
                    self._permission_index[perm.id] = set()
                self._permission_index[perm.id].add(group.id)
                logger.debug(f"Indexed permission {perm.id} for group {group.id}")

        except Exception as e:
            logger.error(f"Failed to add group {group.id} to cache: {e}")
            raise

    def invalidate_group(self, group_id: str) -> None:
        """
        Invalidate all cache entries for a group AND all parent groups in the nesting hierarchy.
        This fixes the Lattice 2.9 bug where only direct children were invalidated.
        """
        try:
            # Invalidate the group itself
            direct_key = self._compute_cache_key(group_id)
            if direct_key in self._store:
                del self._store[direct_key]
                logger.debug(f"Invalidated direct cache entry for group {group_id}")

            # Traverse up the nested group hierarchy to invalidate all parents
            current_group_id = group_id
            visited: Set[str] = set()
            while current_group_id and current_group_id not in visited:
                visited.add(current_group_id)
                # Get all parent groups that have this group as a child
                parent_ids = [
                    pid for pid, children in self._nested_group_index.items()
                    if current_group_id in children
                ]
                for pid in parent_ids:
                    parent_key = self._compute_cache_key(pid)
                    if parent_key in self._store:
                        del self._store[parent_key]
                        logger.debug(f"Invalidated parent group {pid} for child {current_group_id}")
                    current_group_id = pid

            # Invalidate all permissions associated with the group
            for perm_id, group_ids in list(self._permission_index.items()):
                if group_id in group_ids:
                    perm_key = self._compute_cache_key(group_id, perm_id)
                    if perm_key in self._store:
                        del self._store[perm_key]
                        logger.debug(f"Invalidated permission {perm_id} for group {group_id}")
                    group_ids.remove(group_id)
                    if not group_ids:
                        del self._permission_index[perm_id]

            logger.info(f"Successfully invalidated group {group_id} and all nested parents")

        except Exception as e:
            logger.error(f"Failed to invalidate group {group_id}: {e}")
            raise

    def get_group(self, group_id: str) -> Optional[RBACGroup]:
        """Retrieve a group from cache, respecting TTL"""
        cache_key = self._compute_cache_key(group_id)
        if cache_key not in self._store:
            logger.debug(f"Cache miss for group {group_id}")
            return None
        entry = self._store[cache_key]
        if time.time() > entry.expires_at:
            del self._store[cache_key]
            logger.debug(f"Cache entry for group {group_id} expired, removed")
            return None
        logger.debug(f"Cache hit for group {group_id}")
        return entry.value

    def clear(self) -> None:
        """Clear all cache entries and indexes"""
        super().clear()
        self._nested_group_index.clear()
        self._permission_index.clear()
        logger.info("Cleared all cache entries and indexes")

import csv
import logging
import statistics
import time
from dataclasses import dataclass
from typing import Dict, List, Tuple
from lattice_sdk import LatticeClient, PromotionWorkflow, WorkflowStatus
from lattice_sdk.cache import PatchedRBACCache

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class BenchmarkResult:
    version: str
    total_workflows: int
    successful_approvals: int
    failed_approvals: int
    p50_latency_ms: float
    p99_latency_ms: float
    cache_miss_rate: float

def run_lattice_benchmark(
    lattice_version: str,
    num_workflows: int = 1000,
    num_approvers_per_workflow: int = 2
) -> BenchmarkResult:
    """
    Benchmark promotion workflow approval throughput for a given Lattice version.
    Measures latency, success rate, and cache performance.
    """
    try:
        # Initialize client with version-specific config
        client_kwargs = {
            "api_key": "benchmark-key",
            "version": lattice_version,
            "enable_rbac_cache": True
        }
        # Use patched cache for 2.9.3+, default cache for 2.9.0
        if lattice_version >= "2.9.3":
            client_kwargs["cache_backend"] = PatchedRBACCache(ttl=300)
            logger.info(f"Using patched RBAC cache for Lattice {lattice_version}")
        else:
            logger.info(f"Using default RBAC cache for Lattice {lattice_version}")

        client = LatticeClient(**client_kwargs)
        logger.info(f"Starting benchmark for Lattice {lattice_version} with {num_workflows} workflows")

        latencies: List[float] = []
        successful = 0
        failed = 0
        cache_misses = 0
        cache_hits = 0

        for i in range(num_workflows):
            workflow_id = f"bench-workflow-{i}"
            # Create workflow with nested group (matches production setup)
            workflow = PromotionWorkflow(
                workflow_id=workflow_id,
                candidate_id=f"candidate-{i}",
                target_role="Staff Engineer",
                approval_group="eng-backend-staff",  # Nested 4 levels deep
                required_approvers=num_approvers_per_workflow
            )
            try:
                client.workflows.create(workflow)
            except Exception as e:
                logger.error(f"Failed to create workflow {workflow_id}: {e}")
                failed += 1
                continue

            # Submit approvals and measure latency
            for j in range(num_approvers_per_workflow):
                start_time = time.perf_counter()
                try:
                    event = {
                        "event_id": f"bench-evt-{i}-{j}",
                        "approver_id": f"approver-{j}",
                        "workflow_id": workflow_id
                    }
                    client.workflows.submit_approval(event)
                    end_time = time.perf_counter()
                    latency_ms = (end_time - start_time) * 1000
                    latencies.append(latency_ms)

                    # Check cache hit/miss (simplified for benchmark)
                    if hasattr(client.cache, "last_access_was_miss"):
                        if client.cache.last_access_was_miss:
                            cache_misses += 1
                        else:
                            cache_hits += 1

                except Exception as e:
                    logger.error(f"Approval failed for workflow {workflow_id}: {e}")
                    failed += 1
                    break
            else:
                # Verify workflow approved
                status = client.workflows.get_status(workflow_id)
                if status == WorkflowStatus.APPROVED:
                    successful += 1
                else:
                    failed += 1

        # Calculate benchmark metrics
        p50 = statistics.median(latencies) if latencies else 0.0
        p99 = statistics.quantiles(latencies, n=100)[98] if len(latencies) >= 100 else (max(latencies) if latencies else 0.0)
        total_cache = cache_misses + cache_hits
        cache_miss_rate = (cache_misses / total_cache) * 100 if total_cache > 0 else 0.0

        return BenchmarkResult(
            version=lattice_version,
            total_workflows=num_workflows,
            successful_approvals=successful,
            failed_approvals=failed,
            p50_latency_ms=p50,
            p99_latency_ms=p99,
            cache_miss_rate=cache_miss_rate
        )

    except Exception as e:
        logger.critical(f"Benchmark failed for Lattice {lattice_version}: {e}")
        raise

def main() -> None:
    """Run benchmarks for Lattice 2.9.0 and 2.9.3, output to CSV"""
    versions = ["2.9.0", "2.9.3"]
    results: List[BenchmarkResult] = []

    for version in versions:
        result = run_lattice_benchmark(lattice_version=version)
        results.append(result)
        logger.info(f"Completed benchmark for Lattice {version}")

    # Write results to CSV
    with open("lattice_benchmark_results.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow([
            "Version", "Total Workflows", "Successful", "Failed",
            "P50 Latency (ms)", "P99 Latency (ms)", "Cache Miss Rate (%)"
        ])
        for res in results:
            writer.writerow([
                res.version, res.total_workflows, res.successful_approvals,
                res.failed_approvals, f"{res.p50_latency_ms:.2f}",
                f"{res.p99_latency_ms:.2f}", f"{res.cache_miss_rate:.2f}"
            ])
    logger.info("Benchmark results written to lattice_benchmark_results.csv")

if __name__ == "__main__":
    main()

# GitHub Actions workflow to pin Lattice versions and run RBAC tests
name: Lattice Dependency Check
on:
  pull_request:
    paths:
      - "requirements.txt"
      - "pyproject.toml"
jobs:
  check-lattice-version:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Check pinned Lattice version
        run: |
          LATTICE_VERSION=$(grep "lattice-sdk" requirements.txt | cut -d'=' -f3)
          if [[ ! $LATTICE_VERSION =~ ^2\\.9\\.[0-9]+$ ]]; then
            echo "Error: Lattice must be pinned to exact patch version, got $LATTICE_VERSION"
            exit 1
          fi
      - name: Run RBAC nested group tests
        run: pytest tests/rbac/test_nested_groups.py -v --cov=lattice_sdk.rbac

# Pytest fixture for 4-level nested RBAC groups
import pytest
from lattice_sdk import RBACGroup

@pytest.fixture(scope="module")
def nested_rbac_groups():
    """Create 4-level nested RBAC groups matching production hierarchy"""
    top = RBACGroup(id="eng-all", name="Engineering", parent_id=None)
    level1 = RBACGroup(id="eng-backend", name="Backend", parent_id=top.id)
    level2 = RBACGroup(id="eng-backend-senior", name="Senior Backend", parent_id=level1.id)
    level3 = RBACGroup(id="eng-backend-staff", name="Staff Backend", parent_id=level2.id)
    return {
        "top": top,
        "level1": level1,
        "level2": level2,
        "level3": level3
    }

def test_nested_group_workflow_approval(nested_rbac_groups, lattice_client):
    """Test promotion workflow approval for deepest nested group"""
    workflow = lattice_client.workflows.create(
        candidate_id="test-user",
        target_role="Staff Engineer",
        approval_group=nested_rbac_groups["level3"].id
    )
    # Submit required approvals
    for i in range(2):
        lattice_client.workflows.submit_approval(
            workflow_id=workflow.id,
            approver_id=f"approver-{i}"
        )
    assert lattice_client.workflows.get_status(workflow.id) == "APPROVED"

# Prometheus metric exporter for Lattice workflow success
from prometheus_client import Counter, Gauge, start_http_server
from lattice_sdk import LatticeClient

WORKFLOW_SUCCESS = Counter(
    "lattice_workflow_success_total",
    "Total successful promotion workflows",
    ["version", "approval_group"]
)
WORKFLOW_FAILURE = Counter(
    "lattice_workflow_failure_total",
    "Total failed promotion workflows",
    ["version", "approval_group", "reason"]
)
STUCK_WORKFLOWS = Gauge(
    "lattice_stuck_workflows",
    "Number of workflows pending >24 hours",
    ["version"]
)

def export_lattice_metrics():
    client = LatticeClient(version="2.9.3")
    # Scrape stuck workflows
    stuck = client.workflows.list(status="PENDING", older_than_hours=24)
    STUCK_WORKFLOWS.labels(version="2.9.3").set(len(stuck))
    # Scrape success/failure counts (simplified)
    # In production, use Lattice webhooks to update counters in real time
    start_http_server(8000)
    print("Prometheus metrics server running on :8000")