Class TransportTestCounter

java.lang.Object
ai.pipestream.module.pipelineprobe.TransportTestCounter
@ApplicationScoped public class TransportTestCounter extends Object
Per-account, per-step document counter for transport + e2e validation. Shared between TestProcessor (increments) and the e2e/transport services (read).

Counters key on (accountId, stepName): the step_name comes from request.metadata.pipe_step_name, which the engine sets to the graph node_id of the receiving step. Transport tests run a single sidecar terminal node so all observations land under one stepName; e2e tap mode adds more node_ids (e.g., "tap-graph", "tap-sink-success"), so each fan-out point gets its own bucket without clobbering the others.

Three numbers per (account, step):

  • total — every doc received, counting duplicates.
  • unique — distinct doc_ids; a retried doc carries the same id, so unique stays put while total advances. total - unique is the at-least-once reprocessing tax.
  • duplicate — count of doc_ids seen more than once.

Aggregate methods (no stepName arg) sum across every step observed for that account — the legacy transport-test path keeps working unchanged.

  • Constructor Details

    • TransportTestCounter

      public TransportTestCounter()
      Creates the transport-test counter. State is the in-memory per-account counter map; instances are managed by CDI.

  • Method Details

    • increment

      public void increment(String accountId, String stepName, String docId)
      Increment for a (account, step) pair. stepName should be the pipe_step_name on the receiving request — typically the graph node_id.
      Parameters:
      accountId - the account whose counters are updated
      stepName - the step bucket key (pipe_step_name / graph node_id)
      docId - the document id received; tracked for unique/duplicate detection

    • increment

      public void increment(String accountId, String docId)
      Backwards-compatible increment without a step dimension. Logs under the synthetic step name "_default" so legacy callers (direct test invocations that bypass the engine and have no pipe_step_name) continue to count.
      Parameters:
      accountId - the account whose counters are updated
      docId - the document id received

    • getCount

      public int getCount(String accountId, String stepName)
      Total receipts for one (account, step), including duplicates.
      Parameters:
      accountId - the account to query
      stepName - the step bucket to query
      Returns:
      the total number of receipts, or 0 if the bucket does not exist

    • getUniqueCount

      public int getUniqueCount(String accountId, String stepName)
      Distinct doc_ids for one (account, step).
      Parameters:
      accountId - the account to query
      stepName - the step bucket to query
      Returns:
      the number of distinct doc_ids, or 0 if the bucket does not exist

    • getDuplicateCount

      public int getDuplicateCount(String accountId, String stepName)
      Number of duplicated doc_ids for one (account, step).
      Parameters:
      accountId - the account to query
      stepName - the step bucket to query
      Returns:
      the count of doc_ids seen more than once, or 0 if the bucket does not exist

    • getDuplicateDocIds

      public List<String> getDuplicateDocIds(String accountId, String stepName)
      Snapshot of duplicated doc_ids for one (account, step).
      Parameters:
      accountId - the account to query
      stepName - the step bucket to query
      Returns:
      a copy of the duplicated doc_ids, or an empty list if the bucket does not exist

    • getReceivedDocIds

      public Set<String> getReceivedDocIds(String accountId, String stepName)
      Snapshot of every distinct doc_id received at one (account, step). The pipeline-crawl orchestrator's RunLedger pulls this every tick to populate the per-stage receive set, then diffs against the connector's dispatched-ids set to identify the actual missing source pks. Set return type because the underlying storage is a set and duplicates are not meaningful here.
      Parameters:
      accountId - the account to query
      stepName - the step bucket to query
      Returns:
      a copy of the distinct doc_ids received, or an empty set if the bucket does not exist

    • getCount

      public int getCount(String accountId)
      Total receipts across every step observed for this account.
      Parameters:
      accountId - the account to query
      Returns:
      the summed total over all step buckets, or 0 if the account is unknown

    • getUniqueCount

      public int getUniqueCount(String accountId)
      Distinct doc_ids across every step observed for this account.
      Parameters:
      accountId - the account to query
      Returns:
      the size of the union of distinct doc_ids over all step buckets, or 0 if the account is unknown

    • getDuplicateCount

      public int getDuplicateCount(String accountId)
      Total duplicated doc_ids across every step observed for this account.
      Parameters:
      accountId - the account to query
      Returns:
      the size of the union of duplicated doc_ids over all step buckets, or 0 if the account is unknown

    • getDuplicateDocIds

      public List<String> getDuplicateDocIds(String accountId)
      Snapshot of doc_ids duplicated at any step for this account.
      Parameters:
      accountId - the account to query
      Returns:
      a list of the union of duplicated doc_ids over all step buckets, or an empty list if the account is unknown

    • getStepNames

      public Set<String> getStepNames(String accountId)
      Step names observed for this account (in insertion order).
      Parameters:
      accountId - the account to query
      Returns:
      a copy of the observed step-name keys, or an empty set if the account is unknown

    • reset

      public void reset(String accountId)
      Reset all steps for one account.
      Parameters:
      accountId - the account whose counters are removed

    • resetAll

      public void resetAll()
      Reset every account.

    • snapshotByAccount

      public Map<String, Map<String, TransportTestCounter.StepCounts>> snapshotByAccount()
      Snapshot of all in-memory counters keyed by account then step name. Empty when no pipeline runs have completed yet in this process.
      Returns:
      a per-account, per-step map of TransportTestCounter.StepCounts snapshots