feat(ingestion): per-step retry and concurrentlyPerGroup builder API

[pl]

Problem

Two readability problems in the ingestion pipeline framework make pipeline definitions harder to scan than they should be:

1. Grouped concurrent processing takes three chained calls to say one thing: groupBy(fn).concurrently((group) => group.sequentially((b) => ...)). The intermediate builders each expose exactly one method, so the chain forces boilerplate without adding information at the top level.
2. Retry is a special wrapper vocabulary: builder.retry(callback, options) wrapping a whole sub-pipeline, plus a separate pipeBatchWithRetry(step, options). Two production pipelines wrapped long step sequences in a single retry block, so a transient failure late in the chain re-ran every step before it, including pure ones, which is bad for idempotency.

The living docs in nodejs/src/ingestion/framework/docs/ had also drifted: chapter 3 claimed concurrency is always unbounded (there is a maxConcurrency cap now), and several framework features had no chapter at all.

Changes

concurrentlyPerGroup

One top-level call instead of a groupBy().concurrently() chain, same runtime behavior (ConcurrentlyGroupingBatchPipeline is unchanged). The callback receives a group builder whose only method is sequentially, keeping within-group ordering explicit at the call site:

- .groupBy((element) => `${element.team.teamId}:${element.headers.session_id}`)
- .concurrently(
-     (group) =>
-         group.sequentially((b) =>
-             b.retry((rb) => rb.pipe(createResolveKeyStep(keyStore)), { name: 'resolve_session_key', tries: 3, sleepMs: 100 })
-         ),
-     { maxConcurrency: sessionKeyResolutionMaxConcurrency }
- )
+ .concurrentlyPerGroup(
+     (element) => `${element.team.teamId}:${element.headers.session_id}`,
+     (group) =>
+         group.sequentially((b) =>
+             b.pipe(createResolveKeyStep(keyStore), { retry: { name: 'resolve_session_key', tries: 3, sleepMs: 100 } })
+         ),
+     { maxConcurrency: sessionKeyResolutionMaxConcurrency }
+ )

Groups process concurrently (optionally capped), items within a group sequentially. GroupingBatchPipelineBuilder is deleted; GroupProcessingBuilder remains as the group builder type. It holds a pipeline-completion closure rather than the grouping function and previous pipeline as fields: a groupingFn field is contravariant in TOutput and would make BatchPipelineBuilder invariant, breaking covariant subpipeline callbacks such as the identity callback in newBatchingPipeline (there is a why-comment on the constructor).

Retry as a per-step option

pipe(step, { retry }) and pipeBatch(step, { retry }) replace builder.retry(...) and pipeBatchWithRetry(...). A shared RetryOptions { name?, tries?, sleepMs? } lives in the new framework/retry.ts (withStepRetry + withBatchRetry); RetryingPipeline and batch-retry.ts are deleted. Retry semantics per step are unchanged: retriable errors back off and retry, isRetriable === false goes to DLQ, unknown errors rethrow after exhausting tries, attempts observed via ingestion_pipeline_retry_attempts. Failure logs use the same resolved name as the metric label, so a metric spike greps straight to its log lines.

Retry now wraps a single step by design. The two multi-step retry blocks were unwrapped and each step classified by whether it does transient-failure-prone I/O:

ai/pipeline.ts (was one block of 11 steps, tries: 5):

Step	Retry	Reason
hog transform (hog_transform_event)	yes	awaits hog transformer, produces messages
read-only process groups (readonly_process_groups)	yes	group-type fetch (Postgres)
emit event (emit_event)	yes	Kafka produce (see note)
normalize, process AI event, strip person props, prepare, create event, split, record lag	no	pure, no awaits

Per-distinct-id branches (event-subpipeline.ts, ai-event-subpipeline.ts; was one block around the whole branching, tries: 5):

Step	Retry	Reason
hog transform (hog_transform_event)	yes	hog transformer I/O
process personless (process_personless)	yes	can hit Postgres via persons store (mostly cache-backed, kept conservatively)
process persons (process_persons)	yes	person merges and writes
process groups (process_groups)	yes	group-type manager and group store
emit event (emit_event)	yes	Kafka produce (see note)
normalize, prepare, create event, split, record lag	no	pure, no awaits

Note

createEmitEventStep does not await the produce, it pushes the promise into side effects. Its retry therefore only catches synchronous produce throws (for example local queue full); async delivery failures still surface through side-effect handling. Kept for parity with the old blocks.

Warning

Behavior changes: a retriable failure mid-chain no longer re-runs earlier steps; non-retriable errors DLQ at the failing step instead of the whole block; the ingestion_pipeline_retry_attempts metric's name label moves from block names (ai_event, per_distinct_id) to per-step names. Error classification is part of the retry option: a step without { retry } that throws an isRetriable: false error now crashes the consumer instead of DLQing (the old blocks classified errors for every step inside them, pure ones included). In practice only dependency-touching steps throw classified errors, and those all kept retry; ingestion-consumer.test.ts now pins both sides of this (DLQ via a retry-wrapped step, crash via an unwrapped one). The DLQ dlq_step header now names the failing step instead of the last step before the retry block.

Removed pipeConcurrently

Deleted from the builder surface: it duplicated concurrently() (same ConcurrentBatchProcessingPipeline underneath) but took a pre-built Pipeline instead of a builder callback, and had zero production call sites. Its only consumer, a framework integration test, now composes via concurrently((b) => ...).

Living docs

• Chapters 3, 5, 6, 7, 11, 13 updated for the new APIs; chapter 3 no longer claims concurrency is always unbounded and documents maxConcurrency; chapter 7 documents redirect's awaitAck and the preserveKey default; chapter 11 rewritten around per-step retry with a batch-retry section
• New chapters: 14 (newBatchingPipeline), 15 (consuming pipelines via feed()/next() and BatchPipelineUnwrapper), 16 (tophog step metrics)
• Fixed stale API references and file paths in .claude/agents/ingestion/pipeline-composition-doctor.md and the ingestion-pipeline-doctor-nodejs skill

How did you test this code?

Automated tests only (no manual/stack testing):

• Framework unit + framework integration + living docs tests: 40 suites, 449 tests pass, including the new retry.test.ts covering withStepRetry/withBatchRetry through the builder interface (transient retry, non-retriable DLQ, exhaustion rethrow, defaults, metric outcomes, name preservation)
• Touched pipelines: ai + analytics (827 tests), sessionreplay + ml-mirror (72), errortracking pipeline and cymbal tests pass
• Full src/ingestion + tests/ingestion jest run compared against a clean master baseline in the same environment: identical failure sets (26 infra-dependent integration/e2e suites needing Redis/Postgres/Kafka/S3 fail on both), zero regressions from this diff
• The variance regression in GroupProcessingBuilder was reproduced with tsc (typed field breaks the covariant newBatchingPipeline callback in chapter 14) before landing the closure-based fix
• Swept the tree for leftover references to removed APIs: none

The new retry tests replace retrying-pipeline.test.ts and batch-retry.test.ts; they catch the same regressions (retry classification, DLQ conversion, exhaustion) at the new step-wrapper level, plus the new pass-through behavior of pipe/pipeBatch without a retry option.

Automatic notifications

• Publish to changelog?
• Alert Sales and Marketing teams?

Docs update

Internal framework only, living docs updated in this PR. No posthog.com docs affected (skip-inkeep-docs).

🤖 Agent context

Autonomy: Human-driven (agent-assisted)

• Built with Claude Code. I directed the design; Claude ran the exploration (usage inventory of groupBy/retry/pipeBatchWithRetry call sites, docs coverage audit) and delegated implementation to Opus subagents in phases: framework core, call-site migration with the per-step retry review, and docs.
• Naming: considered groupedConcurrently, concurrentlyInGroups, concurrentlyPerGroup; picked concurrentlyPerGroup because the concurrency unit is the group. A first iteration had the callback receive the per-item pipeline builder directly; I changed it to receive a group builder with only sequentially so within-group ordering stays visible at the call site.
• GroupProcessingBuilder internals went through two iterations: a typed groupingFn field broke covariance (TS2322 in the batching docs chapter), a first fix erased the field type to any, and the final version replaces the fields with a typed pipeline-completion closure so nothing is erased.
• pipeConcurrently removal came out of reviewing the builder surface during the docs gap-fill: it was about to be documented despite having no production users, so it was deleted instead.
• Retry scope decisions were made conservatively: ambiguous steps (process_personless, emit_event) keep retry to stay close to the old block behavior; the tables above are the full classification for review.
• Repo skills consulted: /writing-tests conventions for the ported retry tests.

[greptile-apps]

Comments Outside Diff (2)

1. nodejs/src/ingestion/framework/retry.ts, line 106-111 (link)

   Same metric/log name disconnect as withStepRetry: the log uses step.name but the metric uses the resolved name (which may be options.name). Correcting this keeps the two sources aligned.

2. nodejs/src/ingestion/framework/retry.ts, line 33-126 (link)

   OnceAndOnlyOnce: shared retry body

   withStepRetry and withBatchRetry contain identical logic for attempt counting, retryIfRetriable, histogram recording, isRetriable classification, captureException, and rethrowing. The only structural difference is how the step is called (step(value) vs step(values)) and what is returned on a non-retriable error (dlq(...) vs values.map(() => dlq(...))). Extracting a private helper that accepts "call the step" and "build the dlq payload" callbacks would eliminate the duplication and guarantee that any future change (e.g. adding a warn log on exhaustion, changing the histogram label set) is made once.

   Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

_{Reviews (1): Last reviewed commit: "refactor(ingestion): collapse groupBy ch..." | Re-trigger Greptile}

[trunk-io]

     

View Full Report ↗︎ ⋅ Docs

[jose-sequeira]

Seems to simplify! LGTM

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-07-09 15:12 UTC	Run
prod-us	✅ Deployed	2026-07-09 15:43 UTC	Run
prod-eu	✅ Deployed	2026-07-09 15:40 UTC	Run