fix(transform): do not emit empty payloads from a stale aggregate

[Steven Patterson (stevepatterson5)]

Description

Two related but distinct defects in aggregation. The first is the cause of the 400 The request is not valid JSON responses that LiveRamp/substation#3 made visible.

1. Empty payloads from a stale aggregate (the 400 cause).

internal/aggregate.Add refreshes a.now on every successful add as well as on reset:

if time.Since(a.now) > a.maxDuration { return false }

So duration behaves as an idle timeout, not a batch-age limit. Once an aggregate has sat idle longer than that duration, the next Add returns false. aggregate_to_array treats a false Add as "flush the current batch" — but here the batch is still empty, so aggToArray returns nil and an empty message is emitted. send_http_post then POSTs an empty HTTP body, which LogScale rejects as not valid JSON.

Fixed by skipping the emit when the array is empty, in both the control-flush and mid-stream paths, plus a guard in send_http_post so any auxiliary-transform combination is covered.

2. Blank elements when splitting (separate, smaller).

Splitting on a separator yields blank elements for trailing or repeated separators, and NDJSON files normally end with a trailing newline. These became messages and were joined verbatim. In practice a following transform that sets a key (time_now) rescued the blank into a valid object, so the malformed array was rarely seen — instead the blank reached the sink as a content-free event ({"seceng_pipeline_time":...}), injecting one junk record per blank line. Blank and whitespace-only elements are now skipped in both places; the whitespace check also covers the carriage return left when CRLF data is split on a newline.

Motivation and Context

Measured in the seceng LogScale pipelines: roughly 40 rejected batches per hour on gcp_cloudaudit alone (~10.6% of objects). It presents as intermittent because it depends on instance idle time — with cpu_idle enabled, Cloud Run instances routinely sit idle for over the configured minute, and the first batch after an idle gap emits one empty POST.

This explanation is consistent with everything observed: source objects contain no invalid JSON (5,340/5,340 lines parse), failures are independent of trailing newlines (a confirmed-failing object had none), they are size-independent, and rejections return in ~80 ms versus ~17 s for a successful ingest — an empty body is rejected immediately.

For transparency, two earlier hypotheses were investigated and falsified: that trailing-newline blanks corrupted the array (disproved — time_now rescues them), and that bufio.Scanner truncated long lines (disproved — the cap is 128 MB on GCP, and the largest line was 76,612 bytes).

How Has This Been Tested?

• gofmt clean; go build ./... clean; full suite green across 12 packages under TZ=UTC.
• TestAggregateToArrayStaleBatch (new) — drives an aggregate past its duration and asserts no empty payload is emitted.
• TestAggToArray (new) — trailing blank, interior blank, whitespace-only, all-blank, and normal cases; asserts output is always valid JSON.
• TestAggregateFromString — new cases for trailing and repeated separators.
• Each test was verified to actually catch its bug by reverting only the source file and keeping the test: reverting aggregate_to_array.go yields emitted an empty payload; sinks cannot distinguish this from real data.
• Pre-existing FuzzTestAggToArray and FuzzTestAggregateFromString seeds still pass, including the existing empty-input seed.

Without TZ=UTC, TestTimeToString and TestTransform/time_to_string fail on any non-UTC host. Pre-existing and unrelated; CI and the container build both run UTC.

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

Checklist:

• My code follows the code style of this project.
• My change requires a change to the documentation.
• I have updated the documentation accordingly.

Deliberately out of scope

The underlying aggregate.Add semantics are left alone. Refreshing a.now on every add — making duration an idle timeout rather than a batch-age limit — is arguably the deeper design bug, but changing it would alter batching behaviour for every transform and every caller. Worth a maintainer opinion on whether that should be addressed separately.

Verification after deploy

Unit tests prove the mechanism, not that it was the production cause; that is not provable in a test. The confirmation is the 400 rate dropping to zero once deployed, which should be obvious within an hour given ~40/hour on gcp_cloudaudit.

Tracked in SEC-1695.

🤖 Generated with Claude Code

Summary by CodeRabbit

• Bug Fixes
  • Prevented blank or whitespace-only entries from being included in aggregated JSON arrays.
  • Avoided emitting messages created from trailing separators, repeated separators, or blank lines.
  • Stopped empty aggregates from being flushed or posted over HTTP.
  • Improved stale/idle batch behavior to ensure no empty non-control messages are emitted.
  • Added extra validation to ensure aggregation output remains valid JSON.
• Tests
  • Extended coverage with new edge cases and a stale-batch scenario.

[Snyk Service Account (ops-github-snyk)]

✅ Snyk checks have passed. No issues have been found so far.

Status	Scan Engine	Critical	High	Medium	Low	Total (0)
✅	Open Source Security	0	0	0	0	0 issues
✅	Licenses	0	0	0	0	0 issues
✅	Code Security	0	0	0	0	0 issues

💻 Catch issues earlier using the plugins for VS Code, JetBrains IDEs, Visual Studio, and Eclipse.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 7919a964-2e20-4cb1-a3f5-c52c66c89e60

📥 Commits

Reviewing files that changed from the base of the PR and between 8bc51ed and 1894d4a.

📒 Files selected for processing (1)

• transform/aggregate_to_array.go

🔗 Linked repositories identified

CodeRabbit considers these linked repositories for cross-repo context during reviews:

• liveramp/big_data_core (manual)

🚧 Files skipped from review as they are similar to previous changes (1)

• transform/aggregate_to_array.go

---

Walkthrough

Changes

Aggregation now removes blank or whitespace-only elements, suppresses empty aggregate emissions including stale-batch cases, and prevents HTTP POST requests for empty transformed payloads.

Aggregation empty-output handling

Layer / File(s)	Summary
Filter blank aggregation elements transform/aggregate.go, transform/aggregate_from_string.go, transform/*aggregate*_test.go	Blank split elements are omitted from aggregated JSON arrays, with tests covering trailing separators, repeated separators, whitespace, and JSON validity.
Suppress empty aggregate outputs transform/aggregate_to_array.go, transform/aggregate_to_array_stale_test.go	Empty control flushes are skipped, while stale batches reset and re-add the current message without emitting an empty result.
Skip empty HTTP payloads transform/send_http_post.go	HTTP POST requests are skipped when auxiliary transforms produce zero-length payloads.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title is concise and clearly states the primary fix: suppressing empty payloads from stale aggregates.
Description check	✅ Passed	The description includes all required sections and provides detailed context, motivation, testing, and change type.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

---

_{Comment @coderabbitai help to get the list of available commands.}