orchestrata_ultimate.refactor.codemod.observability.graph_writer

Reference › orchestrata_ultimate › refactor › codemod › observability › graph_writer

GraphWriterObserver records codemod runs in the graph database without slowing the operation. It buffers up to buffer_size records (100 by default) of two kinds: a CodemodRun built from each COMMIT_COMPLETED event and a CodemodPass built from each PASS_COMPLETED event. When a buffer is full the new record is dropped, a warning is logged and a counter increases (dropped_count), so the codemod itself is never blocked. flush writes the buffered records through a BoltSyncCodemodRepository from repository; pending_runs and pending_passes report what is waiting. The record types are in _types. The workspace builds it as one of its default observers.

At a glance

Class diagram

Dependency diagram

API

Classes

GraphWriterObserver

Decorators: traces(spec='110', story='US-110-005', reqs=['REQ-110-026', 'REQ-110-028']) · Line 30

Batched, non-blocking graph database writer.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/observability/graph_writer.py

"""GraphWriterObserver — async writes to graph DB, batched, non-blocking.

SPEC-110 §15.2: Buffers events up to ``buffer_size`` and flushes them to
the graph database via ``BoltSyncCodemodRepository``. On buffer overflow,
events are dropped (non-blocking observer) — the codemod operation must
never stall waiting for a graph write.
"""

from __future__ import annotations

import logging
from collections import deque

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod._types import CodemodPass, CodemodRun
from orchestrata_ultimate.refactor.codemod.observability.events import (
    CodemodEvent,
    CodemodEventType,
)
from orchestrata_ultimate.refactor.codemod.persistence.repository import (
    BoltSyncCodemodRepository,
)

logger = logging.getLogger(__name__)

_DEFAULT_BUFFER_SIZE = 100


@traces(spec="110", story="US-110-005", reqs=["REQ-110-026", "REQ-110-028"])
class GraphWriterObserver:
    """Batched, non-blocking graph database writer.

    Events are buffered up to ``buffer_size`` (default 100). When the buffer
    is full, new events are **dropped** (not blocking the codemod operation).
    A flush writes accumulated ``CodemodRun`` and ``CodemodPass`` records to
    the graph database via ``BoltSyncCodemodRepository``.

    The observer extracts persistence-relevant data from events:
        - ``COMMIT_COMPLETED`` → ``CodemodRun`` record
        - ``PASS_COMPLETED`` → ``CodemodPass`` record
    """

    def __init__(
        self,
        repository: BoltSyncCodemodRepository | None = None,
        buffer_size: int = _DEFAULT_BUFFER_SIZE,
    ) -> None:
        self._repository = repository
        self._buffer_size = buffer_size
        self._runs: deque[CodemodRun] = deque(maxlen=buffer_size)
        self._passes: deque[tuple[str, CodemodPass]] = deque(maxlen=buffer_size)
        self._dropped: int = 0

    def on_event(self, event: CodemodEvent) -> None:
        """Buffer *event* for graph persistence; drop on overflow."""
        et = event.type
        data = event.data
        corr = event.correlation_id

        if et == CodemodEventType.COMMIT_COMPLETED:
            run = CodemodRun(
                run_id=corr,
                project=data.get("project", "unknown"),
                started_at=data.get("started_at", ""),
                completed_at=data.get("completed_at"),
                status=data.get("status", "success"),
                dry_run=data.get("dry_run", False),
                pass_count=data.get("pass_count", 0),
                finding_count=data.get("finding_count", 0),
                file_count=data.get("file_count", 0),
                duration_ms=int(data.get("duration_seconds", 0) * 1000),
            )
            if len(self._runs) >= self._buffer_size:
                self._dropped += 1
                logger.warning(
                    "GraphWriterObserver: dropped CodemodRun %s (buffer overflow, %d dropped)",
                    corr,
                    self._dropped,
                )
                return
            self._runs.append(run)

        elif et == CodemodEventType.PASS_COMPLETED:
            pass_record = CodemodPass(
                pass_id=f"{corr}:{data.get('pass_name', 'unknown')}",
                run_id=corr,
                pass_name=data.get("pass_name", "unknown"),
                pass_type=data.get("pass_type", "transform"),
                file_count=data.get("file_count", 0),
                duration_ms=int(data.get("duration_seconds", 0) * 1000),
            )
            if len(self._passes) >= self._buffer_size:
                self._dropped += 1
                logger.warning(
                    "GraphWriterObserver: dropped CodemodPass %s (buffer overflow, %d dropped)",
                    pass_record.pass_id,
                    self._dropped,
                )
                return
            self._passes.append((corr, pass_record))

    def flush(self) -> None:
        """Write all buffered records to the graph database.

        Errors are logged and swallowed — graph writes must never block
        or fail the codemod operation (ERR-110-012).
        """
        if self._repository is None:
            logger.warning(
                "GraphWriterObserver: no repository configured, discarding buffer"
            )
            self._runs.clear()
            self._passes.clear()
            return

        while self._runs:
            run = self._runs.popleft()
            try:
                self._repository.save_run(run)
            except Exception:
                logger.exception(
                    "GraphWriterObserver: failed to save CodemodRun %s", run.run_id
                )

        while self._passes:
            run_id, pass_record = self._passes.popleft()
            try:
                self._repository.save_pass(run_id, pass_record)
            except Exception:
                logger.exception(
                    "GraphWriterObserver: failed to save CodemodPass %s",
                    pass_record.pass_id,
                )

    @property
    def dropped_count(self) -> int:
        """Number of events dropped due to buffer overflow."""
        return self._dropped

    @property
    def pending_runs(self) -> int:
        """Number of CodemodRun records awaiting flush."""
        return len(self._runs)

    @property
    def pending_passes(self) -> int:
        """Number of CodemodPass records awaiting flush."""
        return len(self._passes)