Class RunLedger

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.RunLedger
public final class RunLedger extends Object
Single source of truth for one pipeline-crawl run's progress.

Two halves:

  • Dispatch view (the connector's claim of what it sent): dispatchedIds, expectedRowCount, dispatchComplete. Hydrated by JdbcDispatchSubscriber from StreamCrawlStatus events. Doc ids in dispatchedIds are the connector- authoritative ground truth for "what should have been processed".
  • Receive view (per main-flow stage's tap): stages — one StageLedger per tap node id (tap-chunker, tap-embedder, ...). Hydrated each poll tick from TransportTestCounter.getReceivedDocIds, which the TestProcessorServiceImpl populates as docs land at each tap.

"Done" for the run = dispatchComplete AND every stage's receivedCount >= expectedCount AND no stage has missing ids.

Thread-safe: all mutating fields are atomic / concurrent collections. Stage map is mutated only from the orchestrator's setup path (which runs single-threaded before the subscription + poll loop start).

  • Constructor Details

    • RunLedger

      public RunLedger(String triggerId)
      Creates a ledger for one pipeline-crawl run, stamping the start time as now. Both views start empty and expectedRowCount starts at -1 (not yet announced).
      Parameters:
      triggerId - a non-empty identifier for the run
      Throws:
      IllegalArgumentException - if triggerId is null or empty

  • Method Details

    • forKnownDocs

      public static RunLedger forKnownDocs(String triggerId, Collection<String> dispatchedDocIds)
      Pre-populate a ledger for runs that know their doc set up front (e.g., SINGLE_DOC source — the dispatcher uploads exactly one doc and gets back its id). Fills the dispatch view as if the connector had already emitted EXPECTED_COUNT_KNOWN, every DOC_DISPATCHED, and DISPATCH_COMPLETE.

      The wait loop's isClosedSuccessfully() then returns true once those docs land at every receive-side stage — no subscriber needed.

      Parameters:
      triggerId - any non-empty string identifying the run (e.g., the sidecar-side run id; not a connector trigger id)
      dispatchedDocIds - the exact doc ids the dispatch step emitted; size sets expectedRowCount
      Returns:
      a ledger with its dispatch view fully populated and marked complete
      Throws:
      IllegalArgumentException - if dispatchedDocIds is null or empty

    • triggerId

      public String triggerId()
      The run's trigger id, as supplied at construction.
      Returns:
      the non-empty trigger id

    • startEpochMs

      public long startEpochMs()
      The run's start time, stamped at construction.
      Returns:
      the start time in epoch milliseconds

    • registerStage

      public void registerStage(StageLedger stage)
      Register a receive-side stage. Called once per tap during run setup.
      Parameters:
      stage - the stage ledger to register, keyed by its stage id
      Throws:
      IllegalArgumentException - if stage is null
      IllegalStateException - if a stage with the same id is already registered

    • stage

      public StageLedger stage(String stageId)
      Look up a registered receive-side stage by its id.
      Parameters:
      stageId - the id of a previously registered stage
      Returns:
      the stage ledger for that id
      Throws:
      IllegalStateException - if no stage is registered under stageId

    • stages

      public Map<String, StageLedger> stages()
      A snapshot copy of all registered stages keyed by stage id.
      Returns:
      an immutable copy of the stage-id to stage-ledger map

    • setExpectedRowCount

      public void setExpectedRowCount(int count)
      Set by the connector's EXPECTED_COUNT_KNOWN event. Idempotent if called with the same value; refuses to overwrite once set.
      Parameters:
      count - the expected dispatched row count (must be >= 0)
      Throws:
      IllegalArgumentException - if count is negative
      IllegalStateException - if a different value was already set

    • expectedRowCountOrNegative

      public int expectedRowCountOrNegative()
      The announced expected row count, or -1 if the connector hasn't announced it yet.
      Returns:
      the expected row count, or -1 when not yet announced

    • recordDispatched

      public void recordDispatched(String docId)
      Called per DOC_DISPATCHED event from the connector.
      Parameters:
      docId - the dispatched doc id to record (must be non-empty)
      Throws:
      IllegalArgumentException - if docId is null or empty

    • markDispatchComplete

      public void markDispatchComplete()
      Called once when the connector emits DISPATCH_COMPLETE.

    • dispatchedIdsSnapshot

      public Set<String> dispatchedIdsSnapshot()
      A snapshot copy of the connector's dispatched doc ids.
      Returns:
      an immutable copy of the dispatched-id set

    • dispatchedCount

      public int dispatchedCount()
      The number of doc ids the connector has reported dispatched.
      Returns:
      the current dispatched-id count

    • isDispatchComplete

      public boolean isDispatchComplete()
      Whether the connector has emitted DISPATCH_COMPLETE.
      Returns:
      true once dispatch has been marked complete

    • missingAt

      public Set<String> missingAt(String stageId)
      Doc ids the connector said it dispatched but that haven't landed at the given stage's tap. Empty when the stage's receive set fully covers the connector's dispatched set.
      Parameters:
      stageId - the registered stage to compare the dispatched set against
      Returns:
      an immutable set of dispatched ids not yet received at that stage; empty if dispatch is empty or the stage covers all dispatched ids
      Throws:
      IllegalStateException - if no stage is registered under stageId

    • isClosedSuccessfully

      public boolean isClosedSuccessfully()
      True iff dispatch is complete AND the connector's dispatched-id set is fully covered at every registered stage's receive set. Used by the orchestrator's wait loop as the "done" signal.
      Returns:
      true when dispatch is complete, the expected count is known and equals the dispatched count, and no stage is missing any dispatched id; false otherwise