Search

Search is two services plus a ledger. module-opensearch-sink turns finished documents into indexing requests on a Kafka topic; opensearch-manager is the fleet's only OpenSearch writer and the owner of all index topology; and per-document outcomes flow back as receipts into a Postgres indexing ledger in the repository.

The indexing path

The sink is a terminal pipeline module. Nothing calls it on the data path: a worker loop dials the engine and pulls work units over the ModuleWorkService bidi stream, each carrying an inline PipeDoc. The sink's node config names plan_ids; an IndexPlanCache resolves each plan from opensearch-manager (Caffeine cache, 10-minute TTL, 1000 entries), all-or-nothing — a missing plan, a plan that is not READY, or an empty list is a permanent failure naming every offender. Physical index names are taken only from the plan's server-derived controlled_indexes: topology is referenced, never invented by the sink.

A converter flattens the PipeDoc into a search projection (title, body, tags, source URI/MIME, NLP analysis, quality scores, ownership, embedding vector sets) — the source binary stays in repository storage and never rides Kafka. The publisher then emits one IndexingRequestEvent per plan to the indexing-requests topic. Transport is chosen by serialized size: payloads up to 256 KiB (opensearch-sink.indexing.inline-threshold-bytes) go inline as a google.protobuf.Any; larger ones are saved to repository S3 and travel as a claim-check DocumentReference.

pipestream-protos/repo/proto/ai/pipestream/repository/v1/indexing_receipts.proto:231

message IndexingRequestEvent {
  // Unique request/event identifier (UUID/ULID).
  string event_id = 1;
  // Reference to the saved StreamIndexDocumentsRequest PipeDoc in
  // repository-service. Set when the payload is too large to inline
  // (claim-check path).
  ai.pipestream.data.v1.DocumentReference document_ref = 2;
  // The plan that selected this index.
  string plan_id = 3;
  // Inline payload (StreamIndexDocumentsRequest packed as Any).
  google.protobuf.Any inline_payload = 5;
  // The originating PipeDoc's doc_id. Always set, regardless of
  // transport (inline or claim-check) — it's what the registered
  // UuidKeyExtractor turns into the deterministic Kafka partition
  // key. Same doc_id → same partition → same ordering, every time.
  string document_id = 6;
}

Identity is deterministic throughout: the partition key is UUID.nameUUIDFromBytes(document_id), the event id is uuid("evt:<doc_id>:<plan_id>"), and the request id is uuid("req:<doc_id>:<index_name>") — redeliveries dedup naturally downstream instead of duplicating writes.

One writer, with a brake

Opensearch-manager (unified HTTP+gRPC port 18103) is the only consumer of indexing-requests and the only issuer of _bulk calls in the fleet. Before touching a batch, its consumer blocks on ClusterStrainGauge.awaitHealthy() — a daemon polling _cluster/health for pending-task depth and wait time, with hysteresis: a configurable run of consecutive healthy samples is required to clear the brake, and probe failures keep the last known state, so flaky telemetry can neither wedge indexing nor release an active brake. While the cluster is strained, offsets stay uncommitted and Kafka holds the backlog. Events are then dereferenced concurrently (a virtual thread each), grouped by indexing strategy, and submitted through windowed bulk queues.

opensearch-manager/src/main/resources/application.properties:257-289

bulk-indexing.queue-count=${BULK_QUEUE_COUNT:20}
bulk-indexing.capacity=${BULK_QUEUE_CAPACITY:500}
bulk-indexing.flush-interval-ms=${BULK_FLUSH_INTERVAL_MS:500}
bulk-indexing.max-concurrent-flushes=${BULK_MAX_CONCURRENT_FLUSHES:4}

These numbers were tuned against a live cluster: lowering the flush interval from 2000 to 500 ms dropped per-document wall-clock from ~3.9 s to ~1 s, while raising concurrent flushes from 4 to 16 changed nothing — the ceiling is the cluster's HNSW build rate, ~1.4k chunk-docs/s, not client parallelism (measured 2026-07-10). The chunker's sentences_internal group is ~65% of chunk volume and is excluded from indexing by default (index_internal_chunks=false) partly for this reason. Only rejected_execution item failures retry, with bounded linear backoff; everything else is terminal on first sight. The reason for the single-writer design is stated plainly in the architecture doc: a sink writing its own documents would issue one tiny request per document and give up an order of magnitude of throughput, and a fleet of writers could never share a backpressure signal.

Write outcomes flow back as DocumentIndexedEvent receipts on the indexing-receipts topic — one per input event, carrying per-physical-index sub-outcomes and point-in-time provenance (parser chain, chunker/embedder/semantic configs), so "why does this document have these embeddings?" survives later config mutation. Each receipt carries a ULID attempt_id minted at outcome-materialization time so its lexicographic order reflects delivery order even under emitter backpressure; repository-service folds receipts into the document_index_state ledger with an UPSERT that ignores any attempt id not strictly greater than the stored one. Six outcomes are defined, including PARTIAL_SUCCESS ("at least one physical-index write succeeded, at least one failed terminally... distinct from SUCCESS, which would lie about completeness") and SKIPPED, which is explicitly not a failure but a terminal nothing-to-index state.

Governed IndexPlans

An IndexPlan is a Postgres row — one per base index name — describing name, strategy, vector-set membership, and HNSW/index-setting overrides. Its lifecycle is validate → persist PENDING → provision through exactly one creation path (IndexKnnProvisioner) → flip READY or FAILED with a last_error. The indexing hot path never creates indices or mappings: it calls only requireIndex/requireKnnField and fails loudly. Provisioning drops its own positive caches and probes the cluster live before flipping a plan to READY, and an IndexRecipeMatrixTest pins the triangle invariant: the family a plan declares equals what ListPlanIndices enumerates equals what physically exists, every KNN field at its own embedder's dimension. Calling UpdateIndexPlan on a FAILED plan resets it to PENDING and re-runs the idempotent provisioner — that is also the recovery lever after an out-of-band index deletion.

pipestream-protos/opensearch/proto/ai/pipestream/opensearch/v1/index_plan.proto:139

// HNSW + engine knobs. Every field is `optional` — unset means "use the
// manager's server-side default." Defaults:
//   engine          = "lucene"
//   method_name     = "hnsw"
//   space_type      = "cosinesimil"
//   m               = 16
//   ef_construction = 100
//   ef_search       = 100
message HnswParameters {
  optional string engine = 1;
  optional string method_name = 2;
  optional string space_type = 3;
  optional int32 m = 4;
  optional int32 ef_construction = 5;
  optional int32 ef_search = 6;
}

Higher-level governance RPCs include CreateIndexShape/ MaterializeIndexShape ("design a whole index family in one idempotent call") and ValidatePlanProducibility, which walks a pipeline graph to confirm every vector set a plan expects has an upstream producer. Vector sets themselves are recipes — chunker config, embedding model config, source field, and a source_cel selector expression — whose unique identity includes a hash of the source expression: the same recipe over different text is a different vector set. OSM only length-checks source_cel; there is deliberately no CEL compiler in the manager, because expression semantics are the engine's concern.

Centralized naming

Callers supply only the base name (validated [a-z0-9][a-z0-9_-]*, immutable); IndexNaming derives everything else. Chunk side indices are <base>--chunk--<chunkConfigId>, separate-strategy vector indices are <base>--vs--<chunker>--<embedder>, and chunk-combined KNN columns are em_<embeddingModelId>. The same derivation backs provisioning, the ListPlanIndices read model, and delete fan-out, so the three cannot drift. There is no tenant prefix: account scoping is a document field (ownership.account_id). Name derivation is pinned under hostile locales — a unit test runs it under Turkish and Arabic-Indic locales with Locale.ROOT throughout, and the sink's Gradle test task forwards user.language/ user.country into the forked test JVM.

Three physical indexing strategies

Plans choose one of three physical layouts for vectors, defined in opensearch_document.proto:

• CHUNK_COMBINED — one chunk index per chunk config, with one KNN column em_* per embedder on each chunk row. One BM25 body per chunk, multiple vector lanes alongside it.
• SEPARATE_INDICES — one index per chunker×embedder pair, each with a single vector field. Isolates lanes completely at the cost of more indices.
• NESTED — vs_* nested fields on the parent document.

Admin search and experiments

AdminSearchService.AdminSearch runs keyword (BM25), semantic, or hybrid queries (SEARCH_MODE_KEYWORD | SEMANTIC | HYBRID), fused by reciprocal rank fusion with rank constant 60 or by weighted scores. Semantic arms embed the query server-side via DJL serving using the same model as the lane; an unreachable embedder fails UNAVAILABLE rather than degrading silently. On top of that sit search experiments — CreateSearchExperiment/RunSearchExperiment run one BM25 arm plus one arm per vector set over the same query and return per-arm and fused results — and experiment groups for query-time A/B switching with no redeploy: CreateExperimentGroup, CloneExperimentGroup, RetireExperimentGroup, SetActiveVariant, and ResolveVariant.

opensearch-manager/docs/query-writers-guide.md — executed live 2026-08-06 over a 1000-document court corpus (14k chunks, two embedding lanes)

$ grpcurl -plaintext -d '{"group":"phase3-embedder-shootout"}' localhost:18103 \
    ai.pipestream.opensearch.v1.AdminSearchService/ResolveVariant
{ "label": "control", "active": true,
  "experiment": { "indexName": "idx-pipeline-crawl-pipeline-crawl-e6642ccf--chunk--sentence-10-3",
                  "fusion": "FUSION_METHOD_RRF", "k": 10 },
  "lanes": [ { "fieldName": "em_ALL_MINILM_L6_V2", "knnQueryPath": "em_ALL_MINILM_L6_V2",
               "chunkerConfigId": "sentence-10-3", "dimensions": 384, ... } ] }

A cosine-space KNN query on that lane returned live hits such as 0.7979 | ...:eddb904a | sentence-10-3 | impose but for the employment relationship. As the Court.... The resolved variant tells the caller exactly which index, fusion method, and lanes to query, so switching an A/B group is a server-side flip, not a redeploy.

Traces are for observation, not decisions

Opensearch-manager also hosts the pipeline-events plane: PipelineEventsService projects the engine's StepExecutionRecord audit stream into monthly pipeline-events-<yyyy.MM> indices and serves GetDocumentTrace, FindDocumentTraces (failures sort first), GetPipelineActivity (per-node rollups with p95 durations), and StreamPipelineEvents (history replay then live fan-out, per-subscriber drop-oldest queues). This is the surface that powers per-document debugging UIs.

The doctrine of record is that eventually-consistent telemetry is for observation only: traces are rendered with freshness hints and never gated on. Quoting the rule from the project AGENTS.md (decision of 2026-07-18): "Code must never gate a state transition on trace data ('is doc X indexed yet?' read from the trace is a race by construction)." Decisions ride authoritative stores — lab drain reads the repository indexing ledger via GetCrawlIndexProgress, and replay eligibility comes from server-side verdicts. The ledger read surface is IndexingLedgerService: GetDocumentIndexState, CheckDocumentsIndexed, ListStuckDocuments, GetIndexingProvenance, StreamLedgerRows, and PurgeIndexState.

Failure in traces is a two-axis read: a document rejected by the parser is stored as status=SUCCESS, disposition=REJECT. The measured consequence is recorded in the implementation: 1023 REJECT rows out of 37928 events would never have surfaced as failures if failure were derived from status alone. DROP and SKIPPED are deliberately not failures — they are filtered documents.

Deletes fan out, with a stale-delete guard

Deletion is event-driven too. A consumer subscribes to document-deleted, asks the ledger where the document ever landed, skips any index whose latest successful write postdates the delete (an out-of-order delete cannot erase a newer re-index), and erases the whole family — the parent by _id plus delete-by-query over the --chunk--*/--vs--* siblings. Failures rethrow and dead-letter to document-deleted-dlq: erasure fails closed. A related guard inside the bulk machinery drops a rejected INDEX whose id was successfully DELETEd in the same batch, because retrying it would resurrect the document.

protomolt-search: an experimental distributed Lucene engine

Alongside the OpenSearch plane, the project develops protomolt-search (repository: distributed-search), an experimental Lucene service for streamed, cross-shard vector search. Its distinguishing feature is collaborative HNSW traversal: shards exchange a monotonically increasing global score floor while a query is running, allowing uncompetitive graph paths to stop earlier. The shared-floor HNSW work exists upstream as an Apache Lucene sandbox pull request (#16357); the engine builds on a published fork, ai.pipestream:lucene-*:11.0.0-experimental-SNAPSHOT.

What is implemented today:

• Streaming Lucene kNN search over gRPC and HTTP (HTTP and gRPC share port 48100), with collaborative cross-shard score-floor propagation (knn.collab.KnnNodeService).
• Single-node and ScaleCube-discovered multi-node execution; collection creation, document indexing, deletion, and persisted local shards via ai.pipestream.index.v1.IndexService.
• Server-side embeddings through a configurable DJL endpoint, an optional cross-encoder rerank head on the coordinator, and a plain-Java provider SPI with TEI, OpenVINO/KServe, and model2vec providers plus a cross-provider equivalence harness.
• Schema-as-proto compilation with compatibility classification, and a typed query compiler for text, range, Boolean, vector, and hybrid queries with reciprocal-rank and weighted-linear fusion.

Integration levels differ per surface, and the README says to check before treating an RFC as a live endpoint: the schema compiler and typed query compiler are tested libraries whose v1alpha1 admin and search RPCs are not yet wired; the ai.pipestream.search.v1alpha1 services are proto-and-RFC contract only.

The direction is integration with the platform rather than a second stack: ProtoMolt should own descriptor loading, validation, mapping, compatibility, and protobuf-to-Lucene document projection, while protomolt-search owns shard placement, Lucene lifecycle, distributed execution, and collaborative search. The concrete boundary and migration order are documented in the repository's docs/PROTOMOLT_INTEGRATION.md.

Related

• Repository — the indexing ledger, receipts, and the document store the sink's claim-checks point into.
• Data processing — chunking and embedding produce the vector sets that IndexPlans govern.
• Modules — the demand-pull contract the opensearch sink uses to receive work.
• Document graphs — ValidatePlanProducibility walks these graphs to confirm a plan's producers exist.