chore: avoid marshalling proc error job payloads

[deepshekhardas]

This PR addresses PR rudderlabs#6872

Changes:

• Introduce procErrorJob type to avoid expensive marshal/unmarshal round-trip through EventPayload
• procErrorJob wraps jobsdb.JobT with original parsed events
• Update processor and pipeline_worker to use new type
• Files changed: processor/processor.go, processor/pipeline_worker.go
• 2 files changed, 15 insertions(+), 9 deletions(-)

---

Summary by cubic

Avoids marshaling/unmarshaling processor error job payloads by introducing a procErrorJob wrapper that carries parsed events with jobsdb.JobT. This reduces CPU and allocations in transform and pipeline paths.

• Refactors
  • Add procErrorJob (*jobsdb.JobT + []types.SingularEventT) to bypass EventPayload round-trips.
  • Replace map[string][]*jobsdb.JobT with map[string][]procErrorJob across processor and pipeline worker.
  • Update delivery status recording and user/destination transform flows to use the new type.

^{Written for commit a0889d2. Summary will update on new commits.}

[cubic-dev-ai]

1 issue found across 2 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="processor/processor.go">

<violation number="1" location="processor/processor.go:2600">
P2: `procErrorJob.events` is introduced but never used; delivery-status logic still unmarshals `EventPayload`, so the intended round-trip optimization is not applied.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[cubic-dev-ai]

P2: procErrorJob.events is introduced but never used; delivery-status logic still unmarshals EventPayload, so the intended round-trip optimization is not applied.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At processor/processor.go, line 2600:

<comment>`procErrorJob.events` is introduced but never used; delivery-status logic still unmarshals `EventPayload`, so the intended round-trip optimization is not applied.</comment>

<file context>
@@ -2593,6 +2593,13 @@ func (proc *Handle) destinationTransformStage(partition string, in *userTransfor
+// avoiding an expensive marshal/unmarshal round-trip through EventPayload.
+type procErrorJob struct {
+	*jobsdb.JobT
+	events []types.SingularEventT
+}
+
</file context>