Class StageLedger

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.StageLedger
public final class StageLedger extends Object
One stage's receive-side view in a RunLedger: which doc ids landed at this stage's tap, and how many we expect.

The connector itself has a separate dispatch-side view on the RunLedger top-level (its dispatched set + dispatch-complete flag). This type is for everything downstream of the connector — the tap on each main-flow node ("tap-chunker", "tap-embedder", ...).

Today every stage counts source documents (the chunker emits one PipeStream per source doc with chunks attached as a sub-message). When a future stage starts emitting per-chunk PipeStreams (e.g. a fan-out embedder), the receive-side count there will be in chunks not docs and the cross-stage comparisons in RunLedger need a unit-aware check that refuses to diff a chunk-set against a doc-set. Until that happens, we keep this type unitless rather than carrying decoration that nothing enforces.

Thread-safe.

  • Constructor Details

    • StageLedger

      public StageLedger(String stageId)
      Create a ledger for one stage's tap.
      Parameters:
      stageId - the stage identifier (e.g. "tap-chunker"); must be non-empty
      Throws:
      IllegalArgumentException - if stageId is null or empty

  • Method Details

    • stageId

      public String stageId()
      The stage identifier this ledger tracks.
      Returns:
      the stage id supplied at construction

    • setExpectedCount

      public void setExpectedCount(int count)
      Set the expected receive-count for this stage. Idempotent if called with the same value; refuses to overwrite a previously-set value.
      Parameters:
      count - the expected number of received doc ids; must be >= 0
      Throws:
      IllegalArgumentException - if count is negative
      IllegalStateException - if a different non-negative value was already set

    • expectedCountOrNegative

      public int expectedCountOrNegative()
      The expected receive-count, or -1 if not yet set. Caller treats absence as "we don't know yet."
      Returns:
      the expected count, or -1 if setExpectedCount(int) has not been called

    • replaceReceived

      public void replaceReceived(Set<String> ids)
      Replace the received set in bulk. Used by the orchestrator to rehydrate from TransportTestCounter's authoritative receive set on each poll tick — the test processor populates that counter directly when a doc lands at a tap, so we don't double-count by re-recording here.
      Parameters:
      ids - the authoritative set of received doc ids to snapshot in; copied into an immutable set, so the caller may mutate theirs
      Throws:
      IllegalArgumentException - if ids is null

    • receivedSnapshot

      public Set<String> receivedSnapshot()
      The immutable snapshot of received doc ids visible to this caller. The returned Set is the actual stored reference (already unmodifiable via Set.copyOf(Collection)), so no defensive copy here.
      Returns:
      the current immutable snapshot of received doc ids

    • receivedCount

      public int receivedCount()
      The number of doc ids in the current received snapshot.
      Returns:
      the size of the current received set