Data loss in agg plugin #20

New issue

Open

opened 2026-08-27 15:09:09 +00:00 by erik · 0 comments

erik commented

2026-08-27 15:09:09 +00:00

Owner

For any group that is cut mid-flight, this means the entire remainder of the group is thrown away rather than emitted in a second bucket.

The cut itself is triggered by defaults the operator never set (max-size: 1000, max-real-seconds: 300), and the timer variant is wall-clock driven, so the same input can produce different output depending on how long the process has been running.

Reproduction

Input — 10 events, one key, only the first carrying the start marker:

{"k":"a","n":1,"first":true}
{"k":"a","n":2,"first":false}
…

Config (max-size: 3 stands in for the default 1000 to keep the repro small; the other two triggers are pinned out of reach so the size trigger is the only variable):

pipeline:
  src:
    module: read-file
    config: {path: in.jsonl}
  lines:
    module: parse-lines
    inputs: [src]
  events:
    module: parse-json
    inputs: [lines]
  group:
    module: agg
    inputs: [events]
    config:
      filter: () => true
      key: (e) => e.k
      start: (e) => e.first === true
      max-size: 3
      max-event-seconds: 1000000000
      max-real-seconds: 86400
      ts-field: __none
      view: |
        (events) => ({
          payload: "emitted bucket: n=" +
            events.map((e) => e.n).join(",") + "\n",
        })
  out:
    module: write-file
    inputs: [group]
    config: {path: out.txt}

Observed — 3 of 10 events survive; the process exits 0 with nothing on ERRORS:

emitted bucket: n=1,2,3

Expected — every event accounted for, e.g. 1,2,3 / 4,5,6 / 7,8,9 / 10.

Evidence

Three runs of the same 10 events, varying only how the bucket is cut:

Config	Buckets emitted	Events surviving
start: + max-size: 3	1,2,3	3 of 10
no start:, max-size: 3	1,2,3 / 4,5,6 / 7,8,9 / 10	10 of 10
start: + max-event-seconds: 2	1,2 / 3,4 / 5,6 / 7,8 / 9,10	10 of 10

The third row is the important one: the same plugin already implements the correct behaviour on a different trigger.

Root cause

handle_event has two exits that release a bucket, and they are not consistent.

max-size (and stop) remove the bucket outright — agg.rs:173:

if bucket.len() >= max_size {
    let b = buckets.remove(&key).unwrap();
    self.process_bucket(ctx, &b, output);
    return;
}

max-event-seconds cuts the bucket but re-seeds the key, bypassing start() — agg.rs:196:

let bucket = buckets.get_mut(&key).unwrap();
let last = bucket.pop().unwrap();
let b: Vec<LogEvent> = std::mem::take(bucket);
*buckets.get_mut(&key).unwrap() = vec![last];   // group continues
self.process_bucket(ctx, &b, output);

flush_all — driven by the max-real-seconds ticker and by shutdown — drains every key (agg.rs:235), so after one tick every open group is bucket-less.

The loss then happens here, agg.rs:203:

} else {
    let should_start = match self.start_fn {
        Some(ref start) => start.call_bool(ctx, &event).unwrap_or(false),
        None => true,
    };
    if should_start {
        buckets.insert(key, vec![event]);
    }
    // else: event dropped, silently
}

There is no distinction between "a group ended and a new one has not begun" — where dropping is the intended start/stop semantic — and "a group was cut by a size or time limit and is still in progress" — where dropping is data loss.

Second silent-drop path, same expression

call_bool(...).unwrap_or(false) means a JS error inside start() — a typo, a null deref on an unexpected event shape — also drops the event silently, and drops every subsequent event for that key. Unlike filter, key and view, which all emit to ERRORS on failure, start has no error path at all.

Suggested fix

Treat only stop and end-of-input as terminal. Size and time limits are cuts, and a cut group should continue:

On a max-size release, re-open the key without consulting start() (mirroring what max-event-seconds already does).
Same for flush_all when it is invoked by the max-real-seconds ticker rather than by shutdown — the timer is a flush, not an end of stream.
Give start an error path: emit to ERRORS on a JS exception instead of unwrap_or(false).

A narrower alternative, if changing the semantics is unwelcome: log at warn whenever an event is dropped because start() returned false while a cut was outstanding — turning silent loss into something an operator can see. This is strictly weaker; the group is still lost.

Acceptance criteria

The reproduction above emits every one of the 10 input events across its buckets.
A JS exception in start() surfaces on ERRORS rather than dropping the event.
Regression tests cover: cut-by-size with start: set, cut-by-timer with start: set, and a genuine start/stop cycle where events between groups are still correctly ignored.

Notes

The shipped examples/seim/demo.yml uses start: () => true, which is immune — every event opens a bucket — and is likely why this has not surfaced. Configs using a real start predicate (a request/response pair, a multi-line trace, a session boundary) are exposed, and the exposure grows with runtime because the max-real-seconds timer fires every 5 minutes by default.

Found while building examples/agent-ops/demo.yml; that config omits start: entirely and is not affected.

When agg releases a bucket because of max-size or the max-real-seconds timer, it removes the bucket entirely. The next event for that key then hits the "no bucket yet" branch and must satisfy start() to open a new one. If it doesn't, the event is discarded — no error, no log line, no counter, exit code 0. For any group that is cut mid-flight, this means the entire remainder of the group is thrown away rather than emitted in a second bucket. The cut itself is triggered by defaults the operator never set (max-size: 1000, max-real-seconds: 300), and the timer variant is wall-clock driven, so the same input can produce different output depending on how long the process has been running. Reproduction Input — 10 events, one key, only the first carrying the start marker: ```json {"k":"a","n":1,"first":true} {"k":"a","n":2,"first":false} … ``` Config (max-size: 3 stands in for the default 1000 to keep the repro small; the other two triggers are pinned out of reach so the size trigger is the only variable): ```yaml pipeline: src: module: read-file config: {path: in.jsonl} lines: module: parse-lines inputs: [src] events: module: parse-json inputs: [lines] group: module: agg inputs: [events] config: filter: () => true key: (e) => e.k start: (e) => e.first === true max-size: 3 max-event-seconds: 1000000000 max-real-seconds: 86400 ts-field: __none view: | (events) => ({ payload: "emitted bucket: n=" + events.map((e) => e.n).join(",") + "\n", }) out: module: write-file inputs: [group] config: {path: out.txt} ``` Observed — 3 of 10 events survive; the process exits 0 with nothing on ERRORS: emitted bucket: n=1,2,3 Expected — every event accounted for, e.g. 1,2,3 / 4,5,6 / 7,8,9 / 10. Evidence Three runs of the same 10 events, varying only how the bucket is cut: | Config | Buckets emitted | Events surviving | |:--------------------|:-----------------------|:------------------| | start: + max-size: 3 | 1,2,3 | 3 of 10 | | no start:, max-size: 3 | 1,2,3 / 4,5,6 / 7,8,9 / 10 | 10 of 10 | | start: + max-event-seconds: 2 | 1,2 / 3,4 / 5,6 / 7,8 / 9,10 | 10 of 10 | The third row is the important one: the same plugin already implements the correct behaviour on a different trigger. Root cause handle_event has two exits that release a bucket, and they are not consistent. max-size (and stop) remove the bucket outright — agg.rs:173: ```rust if bucket.len() >= max_size { let b = buckets.remove(&key).unwrap(); self.process_bucket(ctx, &b, output); return; } ``` max-event-seconds cuts the bucket but re-seeds the key, bypassing start() — agg.rs:196: ```rust let bucket = buckets.get_mut(&key).unwrap(); let last = bucket.pop().unwrap(); let b: Vec<LogEvent> = std::mem::take(bucket); *buckets.get_mut(&key).unwrap() = vec![last]; // group continues self.process_bucket(ctx, &b, output); ``` flush_all — driven by the max-real-seconds ticker and by shutdown — drains every key (agg.rs:235), so after one tick every open group is bucket-less. The loss then happens here, agg.rs:203: ```rust } else { let should_start = match self.start_fn { Some(ref start) => start.call_bool(ctx, &event).unwrap_or(false), None => true, }; if should_start { buckets.insert(key, vec![event]); } // else: event dropped, silently } ``` There is no distinction between "a group ended and a new one has not begun" — where dropping is the intended start/stop semantic — and "a group was cut by a size or time limit and is still in progress" — where dropping is data loss. Second silent-drop path, same expression call_bool(...).unwrap_or(false) means a JS error inside start() — a typo, a null deref on an unexpected event shape — also drops the event silently, and drops every subsequent event for that key. Unlike filter, key and view, which all emit to ERRORS on failure, start has no error path at all. Suggested fix Treat only stop and end-of-input as terminal. Size and time limits are cuts, and a cut group should continue: - On a max-size release, re-open the key without consulting start() (mirroring what max-event-seconds already does). - Same for flush_all when it is invoked by the max-real-seconds ticker rather than by shutdown — the timer is a flush, not an end of stream. - Give start an error path: emit to ERRORS on a JS exception instead of unwrap_or(false). A narrower alternative, if changing the semantics is unwelcome: log at warn whenever an event is dropped because start() returned false while a cut was outstanding — turning silent loss into something an operator can see. This is strictly weaker; the group is still lost. Acceptance criteria - The reproduction above emits every one of the 10 input events across its buckets. - A JS exception in start() surfaces on ERRORS rather than dropping the event. - Regression tests cover: cut-by-size with start: set, cut-by-timer with start: set, and a genuine start/stop cycle where events between groups are still correctly ignored. Notes The shipped examples/seim/demo.yml uses start: () => true, which is immune — every event opens a bucket — and is likely why this has not surfaced. Configs using a real start predicate (a request/response pair, a multi-line trace, a session boundary) are exposed, and the exposure grows with runtime because the max-real-seconds timer fires every 5 minutes by default. Found while building examples/agent-ops/demo.yml; that config omits start: entirely and is not affected.

erik added the

priority

high

label

2026-08-27 15:09:09 +00:00

erik self-assigned this

2026-08-27 15:09:09 +00:00