Bellwether.Dedupe (bellwether v0.1.0)

Collapsing repeats before they reach the Spool.

A supervisor in a restart loop emits thousands of identical reports a minute on a link that affords kilobytes an hour. Three defences, in order of how badly they are needed:

One exemplar per Fingerprint per window, carrying a count.
A cap on the number of distinct Fingerprints per window, so a Fingerprint that varies per call cannot flood the Spool by never repeating.

A third sits above this one and outside the Adapter: the Agent honours a rate limit the server returns, so a fleet-wide flood can be shed from the one place that can see it is fleet-wide.

The state is a plain struct and every decision is a pure function of it, so the awkward cases — a window boundary landing mid-loop, the cap being hit by the one crash that mattered — are testable without a running system.

Summary

Types

decision()

t()

Functions

flush(state, now)

Everything held back in the window that just closed, as {fingerprint, count} pairs, ready to be sent as counts rather than copies.

new(opts \\ [])

Options: :window_ms, :max_fingerprints.

observe(state, fingerprint, now)

Decide what to do with an occurrence of fingerprint at now (ms).

over_cap(state)

Occurrences refused because the distinct-Fingerprint cap was reached.

suppressed(state)

How many occurrences this window has swallowed.

Types

decision()

@type decision() :: {:emit, pos_integer()} | :suppress | {:suppress, :fingerprint_cap}

t()

@type t() :: %Bellwether.Dedupe{
  counts: term(),
  emitted: term(),
  max_fingerprints: term(),
  over_cap: term(),
  suppressed: term(),
  window_ms: term(),
  window_start: term()
}

Functions

flush(state, now)

@spec flush(t(), integer()) :: {[{String.t(), pos_integer()}], t()}

Everything held back in the window that just closed, as {fingerprint, count} pairs, ready to be sent as counts rather than copies.

Only Fingerprints seen more than once appear: the first occurrence was already emitted.

new(opts \\ [])

Options: :window_ms, :max_fingerprints.

observe(state, fingerprint, now)

@spec observe(t(), String.t(), integer()) :: {decision(), t()}

Decide what to do with an occurrence of fingerprint at now (ms).

Returns {decision, state}. {:emit, n} means send this one and report that it stands for n occurrences — n is 1 for a first sighting and the accumulated total when a window rolls over.

over_cap(state)

Occurrences refused because the distinct-Fingerprint cap was reached.

Reported as its own number rather than folded into suppressed/1: hitting the cap means Fingerprints are not repeating, which is a different fault from a loop and usually means a Fingerprint is carrying data it should not.

suppressed(state)

How many occurrences this window has swallowed.