Class RichObservationCollector

java.lang.Object
ai.pipestream.module.pipelineprobe.RichObservationCollector
@ApplicationScoped public class RichObservationCollector extends Object
Per-(key, step) rich observation collector for the Pipeline-tab panels, fed live by PipelineEventsConsumer from the engine's pipeline-events audit stream. The key is the run's crawl_id. Captures the six-metric Node Outcome model the FE renders:
  • In / Out / No-op / Reject (+reasons) — exclusive per-doc classification.
  • Sliding-window throughput (recent N-second window + frozen overall).
  • Service-time latency histogram (when the feed carries a dispatch stamp).
  • Recent doc-id ring buffer + error / soft-error counts.

Exclusive classification. pipeline-events emits a rich success record AND a separate NO_OP/REJECT disposition record for the same doc, so each (step, doc) is resolved to exactly one outcome with precedence reject > no-op > processed, order-independent. Out = processed; In = all distinct docs at the node = processed + no-op + reject.

Memory bound: buckets are not GC'd until reset(String) clears the key; OrchestrationDriver evicts each crawl_id when its run ends.

  • Constructor Details

    • RichObservationCollector

      public RichObservationCollector()
      Default constructor for CDI.

  • Method Details

    • record

      public void record(String key, String stepName, String docId, long dispatchedAtEpochMs)
      Record a PROCESSED observation (the node forwarded the doc onward).
      Parameters:
      key - run key (the crawl_id)
      stepName - receiving step (e.g. tap-chunker)
      docId - doc id (for in/out/dup tracking)
      dispatchedAtEpochMs - upstream dispatch stamp for latency; pass 0 to drop latency

    • recordError

      public void recordError(String key, String stepName, String docId, String errorMessage)
      Record a failed observation — bumps error count and replaces last_error_message.
      Parameters:
      key - run key (the crawl_id)
      stepName - the step at which the error occurred
      docId - doc id that failed
      errorMessage - the error message; ignored if null or empty

    • recordNoOp

      public void recordNoOp(String key, String stepName, String docId)
      Record a NO-OP outcome (node forwarded the doc unchanged — nothing to do).
      Parameters:
      key - run key (the crawl_id)
      stepName - the step that no-op'd the doc
      docId - doc id that was forwarded unchanged

    • recordReject

      public void recordReject(String key, String stepName, String docId, String reason)
      Record a terminal REJECT for a doc at this step, with the reason name.
      Parameters:
      key - run key (the crawl_id)
      stepName - the step that rejected the doc
      docId - doc id that was rejected
      reason - the reject reason name; UNSPECIFIED is used when null or empty

    • recordSoftError

      public void recordSoftError(String key, String stepName, String docId, String errorMessage)
      Record a non-fatal SOFT ERROR (the doc continued) for a doc at this step.
      Parameters:
      key - run key (the crawl_id)
      stepName - the step at which the soft error occurred
      docId - doc id that hit the soft error
      errorMessage - the error message; ignored if null or empty

    • snapshot

      public RichObservationCollector.Snapshot snapshot(String key, String stepName)
      Snapshot the rich state for one (key, step). Returns null if no observations have been recorded for this pair yet — callers should treat that as "no data, don't render a panel".
      Parameters:
      key - run key (the crawl_id)
      stepName - the step (e.g. tap-chunker)
      Returns:
      the snapshot, or null if nothing recorded

    • getStepNames

      public Set<String> getStepNames(String key)
      Step names with at least one observation for the given key.
      Parameters:
      key - run key (the crawl_id)
      Returns:
      the step names, or an empty set

    • totalRejectedDocs

      public long totalRejectedDocs(String key)
      Distinct docs whose terminal outcome anywhere in the run is REJECT — the run-wide "intentionally dropped" tally (e.g. empty-body docs FILTERED at the parser→chunker edge). Used by completion accounting: a doc the pipeline rejected will never reach the terminal tap, so terminal + rejected is the count to compare against the connector's dispatched total. Union across steps (a doc rejects at exactly one step on its path; the union is dedupe-safe regardless).
      Parameters:
      key - run key (the crawl_id)
      Returns:
      the count of distinct docs with a REJECT outcome anywhere in the run

    • rejectedDocIds

      public Set<String> rejectedDocIds(String key)
      The distinct rejected doc-id set itself (same dedupe semantics as totalRejectedDocs(String)). Completion accounting needs the SET, not just the size: a duplicate redelivery of an already-successful doc can be rejected (HYDRATION_FAILED after the original's transport reclaim), so the rejected set OVERLAPS the terminal set — gating on terminal + rejected >= total double-counts those docs and closes the run early while real work is still queued (observed: the early close's cleanup destroyed 193 still-queued docs of a 4,360-doc crawl). Callers must union the two sets and compare distinct ids.
      Parameters:
      key - run key (the crawl_id)
      Returns:
      distinct doc ids with a REJECT outcome anywhere in the run

    • reset

      public void reset(String key)
      Reset all buckets for one key. Safe to call mid-run.
      Parameters:
      key - run key (the crawl_id) to evict

    • resetAll

      public void resetAll()
      Reset every key.