Engine — Jobs & Orchestration

The engine is the orchestration core of ProtoMolt: it admits documents from an intake stream, walks each one through a versioned graph of processing nodes, and serves work to external modules over gRPC. As its README puts it, it is the one service that owns a document's journey end to end.

It never parses, chunks, or embeds anything itself, and it never calls a module — modules dial in and pull work. Between hops, document bodies live in repository-service (S3) and only small claim-check pointers ride Kafka, so the bus stays tiny regardless of document size. The service is Quarkus/Java on gRPC port 18100; its packages and image names still use the codebase's original pipestream name.

Demand-pull work distribution

The usual orchestration topology is inverted here: the engine never dials a module. For each module service it runs one raw Kafka consumer on a topic named pipestream.module.<service>, buffering small pointer records. A module worker that wants work opens a short-lived bidirectional gRPC stream on ModuleWorkService.Work — one stream per job, on its own virtual thread.

pipestream-protos/module-worker/proto/ai/pipestream/module/work/v1/module_work_service.proto

// Each work unit lives in its own bidi stream:
//   1. Module dials engine, opens stream.
//   2. Module sends Hello{module_id, instance_id}.
//   3. Engine sends WorkUnit (or NoWorkAvailable if buffer empty after ~5s).
//   4. Module processes; sends periodic Heartbeat (~15s) while working.
//   5. Module sends WorkAck{work_unit_id, status, updated_payload}.
//   ...
//   7. Engine sends AckConfirmed; stream closes.
service ModuleWorkService {
  rpc Work(stream WorkRequest) returns (stream WorkResponse);
}

The conversation is Hello → WorkUnit → Heartbeat → WorkAck → AckConfirmed, then the stream closes. If the buffer is empty after about five seconds (noWorkWait), the engine returns NoWorkAvailable and the module can retry or back off. There is no service discovery for modules and no per-module load balancer; scaling out is opening more streams against the same topic, and scaling to zero is safe because the queue is durable — work simply accumulates.

Two details of the contract are deliberate. First, Hello carries only module_id and instance_id; the identity fields (cluster, graph_id, node_id) are explicitly reserved in the proto — identity rides the pointer instead, so one topic serves every graph node bound to that module service. Second, a watchdog enforces liveness: if neither a Heartbeat nor a WorkAck arrives within twice the heartbeat interval (30s by default), the engine closes the stream, which triggers Kafka redelivery through the same uncommitted-offset mechanism. A transport death before WorkAck re-buffers the record at the head of the partition deque with no quarantine cap; a cancel that races a successful ack is caught later as REJECT(DUPLICATE_DELIVERY).

Screening happens at pop speed, not at module reconnect cadence. Duplicate deliveries and poison records are drained in the Hello handler itself — a 38k-record dead-graph backlog that once throttled live graphs at roughly one record per second drains at about 16k/min.

Versioned graph execution

A pipeline is a versioned DAG: logical, UUID-addressed GraphNodes — each bound to a module service such as parser or chunker — connected by GraphEdges carrying an optional CEL condition, a priority, and a transport_type. Graphs are stored in Postgres as full immutable snapshots, with the version auto-incremented per graph_id and exactly one version marked is_active. Activation publishes a GraphUpdateEvent to a graph-updates Kafka topic, and every engine instance rebuilds its in-memory GraphCache from it.

In-flight version pinning (RFC-0003, shipped) keeps a running document on the graph it started with: at admission the document is stamped with metadata.graph_version and runs that exact version to completion even if a newer one activates mid-flight. The cache is keyed (graphId, version) with a 30-minute expireAfterAccess TTL, and eviction is always safe because a miss reloads the pinned version from Postgres. Hard-deleting a version that still has pinned documents in flight is refused with FAILED_PRECONDITION.

The per-hop loop

When a Hello arrives, the engine pops a buffered record and runs preProcess: resolve the node config from the graph cache, hydrate the document (see below), evaluate the node's CEL filter, and apply CEL pre-mappings. After the module's WorkAck, postProcess runs post-mappings, stamps stream metadata (hop count, processing path, a StepExecutionRecord emitted to the pipeline-events topic), evaluates CEL routing conditions into zero or more destinations, and stages each destination according to its edge behavior.

All CEL — edge conditions, node filters, and mapping transforms — is compiled once at graph-cache rebuild (using Projectnessie CEL for Java) and keyed with edge:, node:, and mapping: prefixes, so no expression is parsed or planned on the hot path.

pipestream-protos — pipeline_config_models.proto (GraphEdge.condition)

// Available context variables:
//   - document: The PipeDoc being processed
//   - metadata: StreamMetadata with processing history
//   - context_params: Map of context parameters
// Example expressions:
//   - "document.search_metadata.language == 'en'"
//   - "metadata.context_params['priority'] == 'high'"
//   - "document.search_metadata.content_length > 1000"
string condition = 5;

Edge behaviors: PERSIST, CACHE, MEMORY

Edges are named by the recovery guarantee they need, not by a mechanism. Every hop declares a transport_type on the wire, and the engine maps it to one of three behaviors:

core-services/pipestream-engine — src/main/java/ai/pipestream/engine/work/EdgeBehavior.java

public static EdgeBehavior of(GraphEdge edge) {
    return switch (edge.getTransportType()) {
        case TRANSPORT_TYPE_MEMORY -> MEMORY;
        case TRANSPORT_TYPE_CACHE -> CACHE;
        default -> PERSIST;
    };
}

• PERSIST stages a destination-specific body in repository-service with DocumentReference.keep=true — a durable recovery and replay point. Unspecified transport maps here: it is the safe default.
• CACHE uses the same S3-stage + Kafka-pointer mechanism but with keep=false, so the body is reclaimed once the destination hop settles. The only mechanical difference between CACHE and PERSIST is that flag — retention per edge, not per document.
• MEMORY keeps the document inline in a bounded per-service queue (default capacity 256) inside the engine instance; bytes never touch object storage, which is what right-to-be-forgotten routing requires.

The choice is a real trade-off, and the project publishes measured guidance. On a 1,000-doc corpus, MEMORY ran about 6x faster than PERSIST on light-compute synthetic documents but only about 3% faster on real embedding-bound documents — the 6x was the transport, not the modules. The documented guidance is: when unsure, PERSIST.

Claim-check hydration

Kafka records are pointers only: a PipeStream carrying a document_ref triple (doc_id, graph_address_id, account_id) plus current_node_id. The record key is a deterministic work_unit_id — a hash of cluster, graph, node, topic, partition, and offset — so redeliveries converge on the same identity. Fan-out writes N destination-specific bodies to the repository and puts N pointers on the bus, each staged at the destination's address, so sibling branches never share a body.

Hydration happens in two levels at each hop. Level 1 resolves the document_ref into a PipeDoc from repository-service (or the Redis claim-check for the intake first hop). Level 2 fetches the raw blob bytes only if the module's declared capabilities say it needs them (caps.needsBlobContent()) — parsers yes, chunkers and embedders no.

The memory settlement ledger

MEMORY edges must not let a durable parent commit early. MemoryWorkRegistry keeps a lineage ledger: a Kafka-backed work unit that spawns MEMORY descendants defers its offset commit (settleWhenClear) until every descendant settles — in the registry's own words, a document's origin Kafka offset advances only when it reaches a PERSIST edge or a terminal. If the engine dies first, the origin offset was never committed, so Kafka redelivers and the chain re-runs idempotently: child work-unit ids are deterministic, derived as hash(parent_work_unit_id, destination_index).

Backpressure is bounded and loud, never silently degraded. A full memory queue waits (20s by default) and then fails retryably — it does not spill to S3, because for an RTBF document writing bytes down is exactly what the policy forbids. A per-graph "MEMORY frontier" occupancy cap (default 5) and a strike ledger bound total exposure; after three strikes (memory-spill-after-attempts, keyed on the deterministic work-unit id so the count survives the redelivery it caused) a hop may duck to CACHE — never to PERSIST — and an RTBF document that strikes out is dropped by design.

Typed module outcomes

A module's reply is a typed outcome, not a bag of bytes. Disposition (CONTINUE/REJECT/RETRY) is orthogonal to whether an error occurred, and WorkAck.result is a oneof with typed reason vocabularies:

pipestream-protos — module_work_service.proto (WorkAck)

oneof result {
  google.protobuf.Any updated_payload = 3;  // CONTINUE: processed
  NoOpResult no_op = 5;                      // CONTINUE: nothing to do
  Reject reject = 6;                         // REJECT: terminal, recorded
}

The no_op branch carries the platform's core invariant, quoted from the proto itself: this is the "never drop, move forward" rule — a module that intends to stop a document must return Reject (recorded, terminal) instead of silently swallowing it. Retryable failures go through the same typing rather than ad-hoc status strings; first-hop intake retries are configured with pipestream.engine.firsthop-retry.max-attempts=5 and a 2s–60s backoff.

The DLQ path

Records that cannot be processed are routed to a per-node dead-letter topic named pipestream.<cluster_id>.<graph_id>.<node_id>.dlq. Each DlqMessage carries the original topic, partition, and offset — full replay coordinates — and the save_on_error option persists the failing body before the DLQ write so the evidence survives. Reprocessing is capped (pipestream.dlq.max-reprocess-count=3).

Diagnostics probes

The engine ships its own diagnostics surface: a gRPC EngineDiagnosticsService — engine-local by design, deliberately not in the shared protos repo, because it is an operational surface rather than a cross-service contract — plus a REST shell under /engine/diagnostics/*. Each probe exercises exactly one named production path, so the probe is the production path, not a shadow of it.

core-services/pipestream-engine — src/main/proto/engine_diagnostics.proto

service EngineDiagnosticsService {
  rpc GetEffectiveConfig(EffectiveConfigRequest) returns (EffectiveConfigResponse);
  rpc ProbeKafka(ProbeKafkaRequest) returns (ProbeResult);
  rpc ProbeClaimCheck(ProbeClaimCheckRequest) returns (ProbeResult);
  rpc ProbeRedis(ProbeRedisRequest) returns (ProbeResult);
  rpc ProbeHop(ProbeHopRequest) returns (ProbeResult);
}

ProbeKafka produces a canary through the real router and consumes it; ProbeClaimCheck stages, hydrates, and purges against the real S3 endpoint; ProbeHop dry-runs a synthetic document through admission, hydration, and routing for a live graph node; GetEffectiveConfig returns resolved values annotated with their source (default vs file vs environment) and secrets redacted. The whole surface is gated by pipestream.engine.diagnostics.enabled (default false) and returns 404 — indistinguishable from absent — when off.

Related