Class IndexPushDriver

java.lang.Object

ai.pipestream.module.pipelineprobe.opensearch.IndexPushDriver

@ApplicationScoped
public class IndexPushDriver
extends Object

Drives court-1000 PipeDocs at the opensearch-sink (OSS) production indexing path: PipeStepProcessorService.processData (unary). TODO: replace when opensearch-sink is demand-pull; engine should route to sink via DAG + claim-check.

This is the testing-sidecar "OpenSearch Index Tests" tab's push button. It bypasses the engine — engine is in flux and not under test here — and calls the same RPC the engine would invoke for an OSS pipeline step.

Why processData, not the experimental StreamDocuments

OSS exposes two services: OpenSearchIngestionService.StreamDocuments (bidi, marked experimental in OSS's own service-registration metadata — delegates to OSM's now-retired bidi StreamIndexDocuments) and the standard PipeStepProcessorService.processData, which is the production path: it iterates the IndexPlans bound to the request's pipeline-step config and XADDs an IndexRequest protobuf to pipestream:indexing:<plan_id> via OpenSearchIndexingPublisher. OSM's RedisIndexingConsumer (PR-1..PR-5) then drains those streams and bulks to OpenSearch. That's the path we want to stress.

Concurrency model

Unary RPC, blocking stub, virtual-thread pump. We spawn one VT per concurrency slot; each pulls from a Semaphore-throttled queue and does a blocking stub.processData(req). The semaphore is the natural backpressure here — every call is a real round-trip — so there's no deadlock risk from the old bidi-batch-flush quirk.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	IndexPushDriver.PushSnapshot	Snapshot of a push job's progress.

Constructor Summary

Constructors

Constructor	Description
IndexPushDriver()	CDI.

Method Summary

Modifier and Type	Method	Description
static Duration	elapsed(IndexPushDriver.PushSnapshot snap)	Compute the duration since job start as a wall-clock helper for the receipts-tail endpoint (deciding whether a job has been quiescent long enough to declare done).
List<IndexPushDriver.PushSnapshot>	listJobs()	Snapshots every tracked job (active and completed), sorted by start time with the most recently started first.
IndexPushDriver.PushSnapshot	snapshot(String jobId)	Look up a job by id.
String	start(List<String> planIds, String opensearchInstance, CourtFixtureSource.Stage stage, int limit, Integer concurrency)	Start a push job.
static Map<String,Object>	toJson(IndexPushDriver.PushSnapshot snap)	Project a snapshot to a UI-friendly map ordered for the tab's progress panel.
static Map<String,Object>	toStartResponse(IndexPushDriver.PushSnapshot snap)	Convenience for a "started" response that callers (e.g.

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

IndexPushDriver

public IndexPushDriver()

CDI.

Method Details

start

public String start(List<String> planIds,
 String opensearchInstance,
 CourtFixtureSource.Stage stage,
 int limit,
 Integer concurrency)

Start a push job. Returns immediately with the job id; the pump runs on virtual threads.

Parameters:

planIds - plan ids OSS will fan out to (REQUIRED — OSS rejects with PROCESSING_OUTCOME_FAILURE "OpenSearch sink requires JSON config carrying plan_ids" if this is empty)

opensearchInstance - Consul service name of the OS cluster to write to; passed as the opensearch_instance field of OpenSearchSinkOptions

stage - which court-1000 corpus to feed

limit - max docs to push; clamped to corpus size (≤ 1000)

concurrency - in-flight cap; null or <= 0 → default; clamped to [1, maxConcurrency]

Returns:

new job id

Throws:

IllegalArgumentException - when planIds is empty, the corpus is missing, or other input is bad

IllegalStateException - when maxJobs active jobs are already running

snapshot

public IndexPushDriver.PushSnapshot snapshot(String jobId)

Look up a job by id.

Parameters:

jobId - job id from start(List, String, CourtFixtureSource.Stage, int, Integer)

Returns:

snapshot or null when no such job

toJson

public static Map<String,Object> toJson(IndexPushDriver.PushSnapshot snap)

Project a snapshot to a UI-friendly map ordered for the tab's progress panel. JSON keys mirror the record field names.

Parameters:

snap - snapshot from snapshot(String)

Returns:

ordered map ready for direct JAX-RS serialization

toStartResponse

public static Map<String,Object> toStartResponse(IndexPushDriver.PushSnapshot snap)

Convenience for a "started" response that callers (e.g. OpenSearchIndexingTestsResource.startPush(String, Map)) return from POST. Matches the shape of toJson(IndexPushDriver.PushSnapshot) so the FE can render the same panel.

Parameters:

snap - snapshot taken right after start(List, String, CourtFixtureSource.Stage, int, Integer)

Returns:

ordered map for direct JAX-RS serialization

listJobs

public List<IndexPushDriver.PushSnapshot> listJobs()

Snapshots every tracked job (active and completed), sorted by start time with the most recently started first.

Returns:

all active and historical jobs, newest-first

elapsed

public static Duration elapsed(IndexPushDriver.PushSnapshot snap)

Compute the duration since job start as a wall-clock helper for the receipts-tail endpoint (deciding whether a job has been quiescent long enough to declare done). Returns Duration.ZERO when the job has not started.

Parameters:

snap - snapshot taken by the caller

Returns:

elapsed wall time since startedAt