Class RunRegistry

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.RunRegistry
@Singleton public class RunRegistry extends Object
Owns the lifetime of pipeline-crawl runs INDEPENDENTLY of any transport connection. Before this registry existed, the SSE resource executed the whole orchestration on the HTTP request's virtual thread — when the client (or the operator's terminal) died, the run died with it, mid-crawl, without cleanup (the leak that produced run ca3eb095 and task #25).

A run started here executes on its own application-owned virtual thread and broadcasts every RunPipelineCrawlResponse to:

  • a bounded in-memory journal (replayable, so a late or re-attaching watcher sees the whole story), and
  • zero or more subscribers (the SSE bridge and/or the gRPC response observer). A subscriber failing or disconnecting detaches that subscriber only — never the run.

What this deliberately does NOT survive: sidecar process death. The journal is in-memory by design (the work queue itself already survives a process death; only the run reporting dies, and the leaked-run sweep is the remedy). Persisting journals was considered and rejected as over-complication for a test harness.

Threading: the orchestration thread is the only writer (one EventEmitter per run, which serializes on the observer — here the broadcast). Replay-then-tail ordering for new subscribers is guaranteed by doing both journal appends and subscribe-replays under the same handle lock.

  • Constructor Details

    • RunRegistry

      public RunRegistry()
      Default constructor for CDI.

  • Method Details

    • start

      public RunRegistry.RunHandle start(ai.pipestream.testing.harness.v1.RunPipelineCrawlRequest request, BiConsumer<ai.pipestream.testing.harness.v1.RunPipelineCrawlRequest, io.grpc.stub.StreamObserver<ai.pipestream.testing.harness.v1.RunPipelineCrawlResponse>> runner)
      Start a run on its own virtual thread and return its handle.
      Parameters:
      request - the crawl request
      runner - the orchestration body (typically PipelineCrawlerImpl::runOrchestration); invoked with the handle's broadcast observer, on the run thread
      Returns:
      the started run's handle

    • get

      public RunRegistry.RunHandle get(String runId)
      Look up a run by id.
      Parameters:
      runId - the run id from event 0 / the runs listing
      Returns:
      the handle, or null if unknown or evicted

    • list

      public List<RunRegistry.RunInfo> list()
      Snapshot of all known runs (live and recently terminal).
      Returns:
      run summaries, newest first