Document Graphs

In ProtoMolt a pipeline is a versioned, self-contained directed acyclic graph of processing steps, stored as a single protobuf message. Every document that travels through the system is pinned to the exact graph version it started on, and every stored byte is qualified by the graph and node that produced it.

A pipeline is a versioned snapshot

A pipeline is one PipelineGraph message. Its GraphNodes and GraphEdges are embedded directly in the message rather than referenced, so each version is a complete, self-contained snapshot: versioning, rollback, and distribution all operate on one value. Each GraphNode is a configured instance of a registered module (module_id plus custom_config), optionally carrying pre/post field mappings, CEL filter conditions, dead-letter configuration, and design-time UI metadata.

core-services/pipestream-protos/config/proto/ai/pipestream/config/v1/pipeline_config_models.proto

message PipelineGraph {
  string graph_id = 1;
  string cluster_id = 2;
  string name = 3;
  string description = 4;
  // Nodes are embedded rather than referenced to ensure graph versions are
  // self-contained snapshots. This simplifies versioning, rollback, and
  // distribution via Kafka (single message contains complete graph state).
  repeated GraphNode nodes = 5;
  repeated GraphEdge edges = 6;
  GraphMode mode = 7;
  ...
  // Version number for optimistic locking.
  int64 version = 10;
  GraphSettings settings = 11;
}

The engine itself is the system of record — there is no separate config service. PipelineGraphService (gRPC) persists full snapshots to Postgres (pipeline_graphs table: UNIQUE(graph_id, version), the graph stored as both JSONB and serialized proto, plus an is_active flag). Only one version is active per graph_id + cluster_id. CreateGraph stages an inactive version; ActivateGraph flips the active pointer. Rollback is activating an older version. Deactivation stops new admissions but never unloads the version — documents pinned to it drain to completion.

Activation distributes the graph

On activation, GraphUpdateEventPublisher pushes the full snapshot to the Kafka topic graph-updates. Every engine instance consumes that topic — in a per-hostname consumer group, so each instance gets its own copy — and rebuilds its local cache. A periodic database reconcile and an on-demand reload-on-miss back this up, so an instance that missed the broadcast still converges.

core-services/pipestream-engine/src/main/resources/application.properties

mp.messaging.outgoing.graph-updates-out.topic=graph-updates
mp.messaging.incoming.graph-updates-in.group.id=pipestream-engine-${HOSTNAME:local}

# Graph drafts (Redis-only; the Postgres graph_drafts table is dropped by V15)
pipestream.engine.drafts.ttl=30d

Clients that want to react to graph lifecycle can open the server-streaming SubscribeToGraphUpdates RPC, which emits CREATED / ACTIVATED / DEACTIVATED / DELETED events filterable by cluster, graph, and type. The designer frontend wraps this as WatchPipelines for its pipeline home screen.

The runtime cache: immutable topologies, pinned versions

Inside the engine, GraphCache is a Caffeine cache keyed by (graphId, version) holding immutable GraphTopology snapshots — nodes indexed by id, edges grouped by from-node — with a separate activeVersions pointer. Activation installs the snapshot in one cache.put and only then flips the pointer, so any single read observes exactly one coherent graph version: it can never combine the nodes of one version with the edges of another.

Loading a graph fires a GraphLoadedEvent, and CelEvaluatorService observes it asynchronously to pre-compile every CEL expression in the graph — edge conditions, node filters, mapping filters, selectors, expressions — into an AST cache. The first document after a deploy does not pay compile latency.

Version pinning (RFC-0003) is the rule that a document finishes on the version it started. At admission, the engine stamps the active version onto the stream (StreamMetadata.graph_id, field 15, and graph_version, field 16), and every hop resolves through getNodePinned(graphId, version, nodeId) against exactly that version — even after a newer version activates. A graph_version of 0 means "unpinned: resolve the active version," used by old streams and diagnostics. Cached versions expire after pipestream.engine.graph-cache.version-ttl (default PT30M) of no access; eviction is safe because a miss transparently reloads that exact version from Postgres.

• Deleting a version that is active or has pinned documents is refused with FAILED_PRECONDITION. There is no force flag: deactivate first and let pinned documents drain.
• Orphaned pins fail loud. A document pinned to a deleted version, or to a node removed from that version, fails at the next reload-miss with telemetry, a DLQ entry, and REJECT(HYDRATION_FAILED), with an actionable message naming the node, graph, and version and how to re-pin. This was motivated by a real 38k-record backlog against a deleted graph.
• Activation races are fenced. GraphCache keeps a per-graph monotonic deactivationEpochs counter; a caller that read the database before a deactivation committed cannot install the now-stale snapshot afterward. The counter is never removed, even on graph delete, because it must stay monotonic.

Identity is graph-qualified

Every stored document row is keyed (doc_id, graph_address_id, account_id, graph_id), where graph_address_id is the node within the graph where that copy of the document lives. Two graphs that both name a hop opensearch-sink can never collide on the same row. Intake is modeled as its own single-node graph per account, intake:<accountId>, so documents that have only been admitted — not yet processed by any pipeline — still have a graph identity.

core-services/pipestream-engine/src/main/java/ai/pipestream/engine/graph/GraphIds.java

public static String intakeGraphId(String accountId) {
    if (accountId == null || accountId.isBlank()) {
        throw new IllegalArgumentException(
                "accountId must not be blank: the intake graph id is account-derived "
                        + "(intake:<accountId>) and blank-graph rows are unrepresentable");
    }
    return INTAKE_GRAPH_PREFIX + accountId;
}

A blank graph_id is unrepresentable by construction: repository-service rejects blank-graph saves, reads, and reclaims with INVALID_ARGUMENT. On the wire, the save contract is an explicit oneof with no inference:

core-services/pipestream-protos/.../pipedoc_service.proto

oneof graph_address {
  bool use_datasource_id = 4;    // an INTAKE save
  string graph_location_id = 5;  // a PIPELINE save at the named graph node
}
// Graph ID that produced this document. REQUIRED on both origin arms ...
// A save without a graph_id is rejected naming the field; blank-graph
// rows are unrepresentable.
optional string graph_id = 8;

The architect's stated mental model is addressing: a graph id is like an IP address, the tuple (graph_id, node_id, doc_id) is the address a document lives at and routes by, replay is re-resolving addresses, and a prefix-delete is dropping a subnet. Consistent with that, account creation provisions an empty cluster by design — the engine never imposes an opinionated default DAG on a new account.

Edges: conditions, transports, and filter outcomes

Each GraphEdge carries an optional CEL routing condition, a priority for ordering competing edges, a transport_type declaring the hop's durability behavior, and max_hops loop protection.

core-services/pipestream-protos/config/proto/ai/pipestream/config/v1/pipeline_config_models.proto

message GraphEdge {
  string edge_id = 1;
  string from_node_id = 2;
  string to_node_id = 3;
  // CEL (Common Expression Language) condition for routing decisions.
  // Example expressions:
  //   - "document.search_metadata.language == 'en'"
  //   - "metadata.context_params['priority'] == 'high'"
  //   - "document.search_metadata.content_length > 1000"
  string condition = 5;
  // Lower values indicate higher priority.
  int32 priority = 6;
  TransportType transport_type = 8;
  // Prevents infinite loops in cyclic graph configurations.
  int32 max_hops = 10;
}

Per-edge transport types

Durability is named per edge, not per pipeline:

• TRANSPORT_TYPE_PERSIST — durable claim-check: the document is staged in object storage and a pointer rides Kafka. Replayable.
• TRANSPORT_TYPE_CACHE — the same staging-plus-pointer mechanics as PERSIST, but the staged copy is reclaimed after consumption. (The name is historical; the enum comment notes it does not create a Redis stream — the wire numbers survive from older transport technology names.)
• TRANSPORT_TYPE_MEMORY — an in-engine inline handoff. Fastest, not replayable.

The transport type declares a durability floor. A right-to-be-forgotten document declares a ceiling, and the ceiling wins: every edge of an RTBF document is forced to MEMORY and no bytes are ever staged.

Filter outcomes: DROP vs SKIP

Node filter conditions have explicit outcomes. FILTER_ACTION_DROP is terminal: it commits the upstream offset and records a typed successful disposition — not an error, no DLQ entry — visible on the event board without a failure receipt. FILTER_ACTION_SKIP skips the module call and routes the document onward unchanged. When no action is specified the default is DROP, which preserves the behavior of graphs written before the field existed.

Validation, subscriptions, and drafts

Mode-tiered validation on deploy

ValidationService.ValidateGraph runs in three modes: DRAFT (structural checks only, no lookups), DESIGN (adds tenancy checks), and PRODUCTION (adds index-plan producibility). Rules are small CDI-discovered GraphRule classes — NoCyclesRule (Johnson's algorithm), CelExpressionRule, EdgeEndpointsRule, OrphanNodeRule, IntakePresenceRule, DuplicateIntakeDatasourceRule, SinkRequiresPlansRule, and others. Deploy always revalidates the exact payload being shipped and refuses with FAILED_PRECONDITION on failure; the CEL validator is built from the same environment definition as the runtime, so an expression the validator accepts is one the runtime compiles.

Datasource subscriptions are derived from the graph

A NODE_TYPE_INTAKE node's datasource_id field is the subscription declaration. On activation the engine derives Subscription rows (keyed {graph_id}:{version}:{datasource_id}) from the graph, so the graph is the source of truth for which datasources feed it, and one datasource can fan out to many graphs, including across accounts. The subscription lookup cache deliberately never caches failures: a database blip during load throws instead of returning an empty list, because a cached empty list is indistinguishable from "no pipeline bound" and intake would ack the record as if it were genuinely unbound.

Drafts are a first-class tier

Between "editing" and "live" there are three draft tiers: browser localStorage autosave, a warm buffer in the BFF (15-minute sliding TTL), and engine-side drafts in Redis (draft:{cluster}:{graph}:{user}, 30-day sliding TTL — every save and every read resets it). Draft saves run DRAFT-mode validation and store a DraftValidation snapshot with a content_hash. A draft also records base_version — which published version it was forked from — because publishing lets the client pick the next version number: a draft forked from v3 that is published after someone else shipped v4 lands as v5 carrying v3's content, and the recorded base is the only evidence that the draft never saw v4. The field is optional int64 specifically so "forked from nothing" is distinguishable from "forked from version 0."

The browser designer

The browser never calls backend gRPC directly. It speaks PipelineDesignerService over connect-es to a Node BFF (port 38106), which translates to the engine's PipelineGraphService. The editor is a Vue 3 + Vue Flow canvas, and the canvas is the model: graphModel.ts maps Vue Flow nodes and edges to PipelineGraph and back, and canvas positions persist in the versioned graph itself via GraphNode.design_config.canvas_x/y, alongside the other DesignModeConfig metadata (UI color and icon, simulated processing time and success rate, sample data). Edge rendering is contract-driven from a single table: MEMORY edges are animated dashed green ("in motion"), CACHE dotted amber, PERSIST solid blue, each with a one-line statement of its tradeoff.

frontend/pipestream-frontend/apps/pipestream-frontend/src/lib/celRecipes.ts

{ id: "english-language",    expression: "document.search_metadata.language == 'en'" },
{ id: "pdf-source",          expression: "document.search_metadata.source_mime_type == 'application/pdf'" },
{ id: "substantial-content", expression: "document.search_metadata.content_length > 100" },
{ id: "processing-not-disabled", expression: "stream.metadata.context_params['skip_processing'] != 'true'" },

Deploy from the designer is one composite BFF call: resolve the next version number, run PRODUCTION-mode preflight validation, materialize index plans via the opensearch-manager, then CreateAndActivateGraph.