Repository — the Document Store

The repository-service is ProtoMolt's document store and claim-check store. It persists PipeDoc bodies to S3-compatible object storage and records one ledger row per stored state in Postgres — its own README puts it as: Postgres is the card catalog, S3 is the stacks.

Every durable pipeline hop, every intake staging copy, and every replay rehydration goes through this one service. The pipeline engine circulates a lightweight DocumentReference over Kafka; the bodies live here. It also owns the indexing ledger — the authoritative record of which documents reached which OpenSearch indices — so "is this document indexed?" is answered from Postgres, never from OpenSearch. The service runs on Quarkus (Java 25), serving gRPC and HTTP on a single port, 18102.

One document, four parts

A stored document is not one blob. PipeDocPartCodec — a pure, IO-free codec — splits a PipeDoc on clean protobuf field boundaries into four parts: CORE, BLOBS, CHUNKS, and PARSED. Each part is written as its own S3 object under a shared {nodeId}/ prefix, and each part object is itself a valid serialized PipeDoc carrying only its own fields plus doc_id. Assembly on read is a field-level mergeFrom of the fragments, round-tripping the original document byte-for-byte — a property the bulk-storage integration tests gate with SHA-256 comparison over a fixture corpus.

CHUNKS is further sub-keyed: one object per chunk set (a consecutive run of semantic_results sharing a result_id), so chunker and embedder outputs from different configurations are independently addressable. Consecutive-run grouping is a byte-fidelity requirement — reassembly must reproduce the exact original ordering, with a #n disambiguator for repeated runs.

The authoritative object list is a PipeDocManifest, stored as a JSONB column on the pipedocs row. Each entry records lifecycle state, size, hash, and provenance:

pipestream-protos/repo/proto/ai/pipestream/repository/pipedoc/v1/pipedoc_parts.proto

// Lifecycle state of one part object.
enum PartState {
  PART_STATE_UNSPECIFIED = 0;
  // Object exists and is readable.
  PART_STATE_PRESENT = 1;
  // Part had no content when written — no object exists.
  PART_STATE_EMPTY = 2;
  // Deliberately deleted (Enhancement B); hash/size retained as provenance.
  PART_STATE_DELETED = 3;
}

message PartManifestEntry {
  PipeDocPart part = 1;
  PartState state = 2;
  int64 size_bytes = 3;            // retained after deletion
  string sha256 = 4;               // retained after deletion
  google.protobuf.Timestamp updated_at = 5;
  string object_key = 6;
  string deleted_reason = 8;       // "POST_PARSE_POLICY" | "RTBF"
  string sub_key = 9;              // CHUNKS only: chunk-set identity
  WriteProvenance written_by = 10; // module, producing node, graph, version
}

Empty parts write no object at all (PART_STATE_EMPTY). The manifest is what makes masked reads, partial saves, retention settlement, and replay eligibility all computable without fetching a single byte from S3.

Reading and writing parts

Masked reads

GetPipeDocRequest.parts and chunk_sets let a caller fetch only the parts it needs — a module that consumes parsed text never downloads the raw source blob. GetPipeDocManifest inspects the part map without fetching any part bytes at all. S3 operations are fanned out concurrently on virtual threads, so a full-document read costs the latency of the slowest part, not the sum. Postgres transactions are scoped to SQL statements only; a connection is never held across an S3 round trip.

Partial saves by server-side copy

Saves are incremental. A writer lists the parts it actually changed in parts_written; every other part is carried forward from a previous stored state via object-storage server-side copy — the bytes never transit the service. The engine derives the written set from observed per-part diffs, not from module declarations.

pipestream-protos/repo/proto/ai/pipestream/repository/pipedoc/v1/pipedoc_service.proto:77-97

// Four-part partial save. Empty = FULL-DOC save: every part is
// written from the supplied pipedoc. Non-empty = write ONLY these
// parts; parts NOT listed are carried forward from
// copy_unwritten_parts_from via object-storage server-side copy
// (their bytes never transit the service).
repeated PipeDocPart parts_written = 11;

// Source address for the parts NOT in parts_written. A partial save
// whose copy source is gone fails FAILED_PRECONDITION.
optional ai.pipestream.data.v1.DocumentReference copy_unwritten_parts_from = 12;

// CHUNKS refinement: the chunk-set sub_keys this save writes.
// Non-empty = write these sub-objects and copy-forward the sibling
// chunk sets like unwritten parts.
repeated string chunk_sets_written = 13;

A copy source that has gone missing — or a source manifest entry that claims PRESENT with a blank object_key — throws CopySourceGoneException and surfaces as FAILED_PRECONDITION, which engages the caller's retry-as-full-save path. Carried-forward copies also work across drives: a partial save can stage a pipeline copy in the platform bucket while carrying parts from a customer-owned intake bucket, using each drive's own credentials.

Intake dedupe

On intake, a SHA-256 checksum match skips the S3 PUT entirely and returns deduplicated=true. A caller that genuinely needs a fresh copy sets force_save=true, which bypasses dedupe and additionally sends the SDK's SHA-256 checksum trailer so the object store itself verifies the landed bytes.

Identity: graph-qualified and deterministic

Every stored state has a node_id — a name-based UUID computed over the document's full coordinates. It doubles as the pipedocs primary key, the S3 key prefix, and a stable Kafka key:

repository-service/src/main/java/ai/pipestream/repository/util/PipeDocUuidGenerator.java:84-85

String composite = docId + SEPARATOR + graphAddressId + SEPARATOR + accountId + SEPARATOR + graphId;
return UUID.nameUUIDFromBytes(composite.getBytes(StandardCharsets.UTF_8));

The graph segment fixes a real incident: two graphs containing a same-named node (for example opensearch-sink) no longer collide in storage. Intake is modeled as a synthetic graph — intake rows carry graph_id = "intake:<accountId>" — and a row_kind discriminator with a database CHECK constraint makes blank identities unrepresentable at rest:

repository-service/src/main/resources/db/migration/V29__intake_graph_identity.sql:36-38

ALTER TABLE pipedocs ADD CONSTRAINT chk_pipedocs_row_kind CHECK (
    (row_kind = 'INTAKE'   AND graph_id LIKE 'intake:%' AND cluster_id IS NULL)
 OR (row_kind = 'PIPELINE' AND graph_id NOT LIKE 'intake:%'));

Determinism means re-ingesting the same document regenerates the same id, so retries converge instead of duplicating state. One deliberate omission: doc_id never appears in S3 keys — for S3 crawls the doc id embeds a source URI whose characters break SigV4 canonicalization against S3-compatible backends. Identity lives in the UUID; the doc_id ↔ object_key mapping lives in the ledger row and inside the .pb bodies themselves.

Two drives per account, bring your own bucket

Every account gets two drives, {accountId}:intake and {accountId}:pipeline, provisioned at account creation via the account-events Kafka consumer. A drive binds to (s3Bucket, s3Prefix, region, credentialsRef, provider), so intake and processing can live in different buckets — even different AWS accounts (provider = CLIENT versus PIPESTREAM). A save for an unprovisioned account hard-fails; the service never falls back to a global bucket. Every save also validates that the account is active via gRPC to the account-manager, behind a Caffeine cache (positive TTL 5 minutes, negative TTL 5 seconds) kept coherent by account events.

Dev and prod differ by configuration only: dev points at a RustFS S3-compatible container with static credentials, prod leaves the endpoint blank (real AWS S3) and uses the default AWS credential chain.

Deletion: lazy in the database, real in S3

Deletes arrive as Kafka events — ReclaimRequest from the engine when an inter-step body is no longer needed, or DocumentDeleted for explicit deletes. The row is tombstoned to PENDING_PURGE, which makes it invisible to reads immediately, and a StoragePurgeEvent is emitted. The BackgroundS3Purger drains that stream batched: one Kafka poll, keys grouped by drive bucket, DeleteObjects up to 1000 keys per round trip, rows finalized in a single transaction sweep. Each row is re-checked under a PESSIMISTIC_WRITE lock before any S3 delete, so a row revived between tombstone and purge voids the event. Purge events are keyed by node id, so the same row can never appear in two consumers' polls at once.

Every hard delete is archived in pipedoc_purges: an append-only purge proof ledger holding identity, hashes, timestamps, and a curated manifest summary, written in the same transaction as the row's finalization. Because node_id is deterministic and therefore not unique over time, the ledger keys on an app-minted surrogate purge_id — an audit ledger must keep every purge of an identity, not just the latest. The table is exempt from identity re-key wipes: the purge doctrine applied to itself.

The S3 write and the ledger row are deliberately not committed together — the store is non-ACID by design. The rule that makes this safe: an S3 object with no live ledger row that owns it is an orphan, and orphans are reclaimable. StorageReconciler enforces it (dry-run by default with a 50-key orphan sample, one-hour minimum age before any delete), and StorageCoherenceDiagnostic exposes the read-only health signal at GET /internal/admin/storage/coherence.

Retention is opt-in: RetentionSweeper expires PIPELINE rows only when pipestream.repository.retention.pipeline is explicitly set — unset means indefinite, and INTAKE rows are never age-swept. Startup refuses to boot on a retention window below the minimum (default P1D), defusing the "bare 60 parses as seconds" footgun that would otherwise expire every row older than a minute.

The indexing ledger

The document_index_state table — with child tables for parser-chain and per-vector-set provenance — is the authoritative record of indexing outcomes. It is fed by consuming DocumentIndexedEvent receipts from the Kafka indexing-receipts topic, and it survives at-least-once delivery and retries through a monotonic UPSERT guarded by a lexicographically ordered ULID:

repository-service/src/main/java/ai/pipestream/repository/indexing/IndexingLedgerWriter.java:76-89

ON CONFLICT (doc_id, plan_id, index_name) DO UPDATE
SET outcome             = EXCLUDED.outcome,
    last_attempt_id     = EXCLUDED.last_attempt_id,
    ...
WHERE EXCLUDED.last_attempt_id > document_index_state.last_attempt_id

The WHERE clause is the load-bearing line: a stale or duplicated receipt is a no-op, so the ledger only ever moves forward.

A companion mechanism settles storage after successful indexing: on a terminal-successful receipt, SourceBlobSettlementService rewrites blobs.pb as a metadata-only husk, flips the manifest entry to PART_STATE_DELETED while retaining its sha256 and size — the map still lists the file, the hash proves what it was — and purges the raw intake .bin. This never happens on FAILED_TERMINAL: a failed document keeps its source for re-parse.

Replay with server-side eligibility verdicts

PipeDocService.StreamDocumentsForReplay (RFC-0004) is a server-streaming query plane for re-running documents through a pipeline. Selectors — doc_ids, crawl_id, datasource_id, time bounds, an indexing-outcome filter — are ANDed together, and a bare account_id is rejected: you cannot accidentally replay an entire tenant. graph_address_id plus graph_id target an interior node's post-hop copy, and required_parts is the engine-computed union of parts the replay target's reachable subtree needs.

Each candidate document comes back with a server-side eligibility verdict — ReplayEligibility: REPLAYABLE, SOURCE_RECEIPT, or BODY_MISSING — computed from its part manifest, so callers never re-derive retention semantics themselves.