Class RunRedisStreamsCleanup

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.RunRedisStreamsCleanup
@ApplicationScoped public class RunRedisStreamsCleanup extends Object
Drops a run's per-graph redis streams once the pipeline-crawl finishes.

Each pipeline-crawl run produces its own per-edge redis streams of the form pipestream:edge:<graphId>:<from>:<to> (and a :poison sibling when delivery overflow fires) plus a per-plan indexing stream pipestream:indexing:plan-<graphId>. The engine's RedisEdgeStreamTransport XDELs each individual entry on successful XACK, so in steady state the streams' XLEN drops to zero — but the empty stream key itself stays in redis along with the consumer-group metadata, and any failed-run leftovers in PEL/poison stay around indefinitely.

Without explicit cleanup, multi-iteration test sweeps accumulate orphaned streams from prior runs (~6GB observed across ~6 runs). This service drops them after each run; the redis ceiling becomes a backstop for true burst headroom rather than the steady-state sink for old data.

Scope: only the streams whose key explicitly carries the supplied graphId. Concurrent runs against different graphs are safe — each call only touches its own keys. UNLINK (rather than DEL) keeps the operation off the main redis thread for the heavy streams.

  • Constructor Details

    • RunRedisStreamsCleanup

      public RunRedisStreamsCleanup()
      Creates the redis streams cleanup service. Instances are managed by CDI.

  • Method Details

    • dropForGraph

      public int dropForGraph(String graphId)
      Drop every per-run stream for graphId. Idempotent — missing keys are silently skipped. Safe to call from a finally block.
      Parameters:
      graphId - the pipeline-crawl graph id; matches the literal segment between pipestream:edge: and the from/to suffixes (and between pipestream:indexing:plan- and end).
      Returns:
      the number of stream keys actually dropped (informational only; not used for control flow).