Documentation
¶
Overview ¶
Package compose provides higher-order monoids that combine simpler monoids:
- Map[K]V — merge two maps by Combining values per key
- Tuple{A,B} — merge two paired values by Combining each component
These let pipelines emit per-event multi-attribute aggregations without separate pipelines per attribute. Example: counting both views *and* unique users per page in one pipeline whose value type is Tuple[int64, []byte] (Sum on the left, HLL on the right).
Index ¶
- func ClampFuture(t, now time.Time, bound time.Duration) time.Time
- func DecayedBytes(amount float64, t time.Time) []byte
- func DecayedBytesNow(amount float64) []byte
- func DecayedSum(halfLife time.Duration, opts ...Option) monoid.Monoid[Decayed]
- func DecayedSumBytes(halfLife time.Duration, opts ...Option) monoid.Monoid[[]byte]
- func DefaultSkewBound(halfLife time.Duration) time.Duration
- func EncodeDecayed(d Decayed) []byte
- func EvaluateAt(d Decayed, halfLife time.Duration, t time.Time) float64
- func MapMerge[K comparable, V any](m monoid.Monoid[V]) monoid.Monoid[map[K]V]
- func TupleMonoid2[A, B any](ma monoid.Monoid[A], mb monoid.Monoid[B]) monoid.Monoid[Tuple2[A, B]]
- type Decayed
- type Option
- type Tuple2
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ClampFuture ¶ added in v0.2.0
ClampFuture pulls a timestamp more than bound ahead of now back to now+bound. Timestamps in the past — however far — are returned untouched, as are all timestamps when bound is non-positive.
Use it in a pipeline whose value extractor takes the timestamp from the EVENT rather than from the clock, which is the only way a future-dated observation can enter the aggregate:
Value(func(e Event) []byte {
ts := compose.ClampFuture(e.OccurredAt, time.Now(), compose.DefaultSkewBound(halfLife))
return compose.DecayedBytes(1, ts)
})
murmur.Trending already stamps every event at its configured clock, so a pipeline built through that preset cannot produce a future frame and does not need this.
Why here and not in Combine ¶
The wall clock is the one piece of information that separates a bogus timestamp from a legitimate one, and it is exactly the piece Combine is not allowed to have: Combine must be pure, or CAS retries and the associativity fuzzer both stop being meaningful (see Combine).
A pairwise bound — clamp the newer operand to the older operand's timestamp plus a limit, using no clock — is the obvious substitute and it does not work. From inside Combine, a state four years in the future beside a real event is indistinguishable from a real event four years after a key went quiet, so any pairwise clamp large enough to be safe for the first case pins legitimately-idle keys at a frame they can never leave: a key idle for 18 hours at a one-minute half-life would come back holding a quarter of its old mass instead of essentially none.
The lift is where an untrusted event timestamp becomes state, it runs once per observation rather than once per merge attempt, and it is the last point at which a clock reading is still honest. So that is where the bound goes.
func DecayedBytes ¶
DecayedBytes lifts a single (amount, time) observation to the wire form expected by DecayedSumBytes pipelines. Pipelines call this from their value extractor:
Value(func(e Event) []byte {
return compose.DecayedBytes(weight(e), e.At)
})
func DecayedBytesNow ¶
DecayedBytesNow is DecayedBytes(amount, time.Now()).
func DecayedSum ¶
DecayedSum returns a monoid that decays older contributions toward the latest timestamp before summing. halfLife controls how fast contributions fade — pass 24*time.Hour for a "last day matters most" feel; 1h for "last hour matters most".
A non-positive halfLife means "no decay": Combine is a plain sum and EvaluateAt returns the stored value unchanged. The two used to disagree — Combine computed 2^(-dt/0), which is NaN at dt=0 and 0 at dt>0, so a zero half-life either poisoned the row with a NaN that survives Encode/Decode or silently dropped every older contribution, while EvaluateAt on the same row reported it undecayed. A zero half-life is reachable by accident: murmur.Trending(name, cfg.HalfLife) with an unset Duration field.
To insert an event at processing time, lift it via DecayedAt(amount, time.Now()). The streaming runtime hands the resulting Decayed value through Combine, the same pattern used by HLL.Single and TopK.SingleN. If the timestamp comes from the EVENT rather than from the clock, run it through ClampFuture first.
func DecayedSumBytes ¶
DecayedSumBytes wraps DecayedSum to operate on []byte values, suitable for plugging into pkg/state/dynamodb.NewBytesStore. Each Combine call decodes both sides, runs the typed Combine, and re-encodes; the cost is a few dozen ns per merge — negligible compared to a DDB round-trip.
The wire format is the same as EncodeDecayed / DecodeDecayed; queries can decode the bytes returned by GetWindow / GetRange and evaluate the score "as of now" via EvaluateAt(d, halfLife, time.Now()).
Pass WithDecodeErrorHandler to be told when an operand is not a Decayed at all; without it the mismatch is recovered from silently, the same deal as the sketch monoids.
func DefaultSkewBound ¶ added in v0.2.0
DefaultSkewBound is the skew allowance ClampFuture uses when a caller has no better number: two half-lives. It is scale-free, and it bounds the damage a skewed observation can do to a factor of four on the contributions that land while the clock catches up.
A half-life past half the Duration range would double into overflow, so it is returned unmultiplied; nobody reaches that by accident.
func EncodeDecayed ¶
EncodeDecayed marshals a Decayed observation to its 17-byte wire form. Identity (Set=false) encodes as zeros, which is intentional — DDB `attribute_not_exists` reads return that shape and DecodeDecayed maps it back to Identity.
func EvaluateAt ¶
EvaluateAt returns the value of d evaluated at time t, decayed forward from d.T. Use this from the query layer when "the value as of now" matters more than the stored reference time.
Evaluating at a t BEFORE d.T returns d.Value unchanged rather than scaling it up. Un-decaying is always an over-estimate — it re-inflates every contribution recorded between t and d.T as though it had been observed at t — and the over-estimate is unbounded: one event stamped a year ahead evaluates to 7.5e109 at halfLife=24h, and one stamped four years ahead to +Inf. A single such row outranks every honest score in the index forever, which is not a defensible reading of "the value as of now". This is also the read-side backstop for a row that was frozen by a future-dated observation before ClampFuture was in the pipeline: the key is stuck, but it is stuck at a finite value rather than at infinity.
Returns 0 for an unset Decayed, and d.Value for a non-positive halfLife.
func MapMerge ¶
MapMerge returns a monoid that merges map[K]V values by Combining matching keys via the inner monoid m, taking the union of the two key sets.
Identity is a nil map; backends that materialize this on first write should treat nil and empty as equivalent. Combine is associative iff the inner monoid m is associative.
func TupleMonoid2 ¶
TupleMonoid2 returns a monoid that merges Tuple2[A,B] componentwise via the inner monoids ma and mb. Useful for pipelines that aggregate multiple metrics in lockstep per key (e.g., view count + unique-visitor HLL).
Types ¶
type Decayed ¶
type Decayed struct {
// Value is the current decayed sum at time T.
Value float64
// T is the reference timestamp (Unix nanoseconds).
T int64
// Set is true when this observation carries a real value; false for Identity.
Set bool
}
Decayed is a (value, time) observation under exponential decay. Combine takes the most recent timestamp's reference frame and decays the older value forward to it before adding. With an appropriate half-life, this implements time-weighted moving sums and averages without windowed bucketing.
Mathematically: Combine((v_a, t_a), (v_b, t_b)) where t_b ≥ t_a is
(v_a * 2^(-(t_b - t_a)/halfLife) + v_b, t_b)
Identity is the unset Decayed; the Set flag distinguishes "no value yet" from a legitimate (0, t=0) observation. This preserves the identity law: Combine(Identity, x) == x for all x.
Associativity is exact in real arithmetic; in IEEE-754 floats it holds within ULP for typical inputs but is not bitwise.
Future-dated timestamps ¶
The reference frame Combine adopts is the newer of the two timestamps, so an observation stamped far ahead of real time freezes the key: 2^(-4y/24h) underflows to exactly zero, the accumulated mass is annihilated, and every real event that follows is itself the older operand and is annihilated in turn. The key sits at whatever the bogus observation carried.
Combine cannot defend against this, and deliberately does not try — see ClampFuture for where the defense lives and why it cannot live here. EvaluateAt contains the blast radius on the read side: it will not scale a frozen value up.
func DecayedAt ¶
DecayedAt builds a Decayed observation for use as a per-event delta. amount is the raw contribution at time t. The returned value has Set=true so it round-trips through Combine(Identity, ...) correctly.
t is honoured exactly. If it came off an event rather than off a clock, wrap it in ClampFuture — Combine will not second-guess it later.
func DecayedNow ¶
DecayedNow is equivalent to DecayedAt(amount, time.Now()).
func DecodeDecayed ¶
DecodeDecayed parses the 17-byte wire form back into a Decayed.
An empty input is the absent key and decodes to Identity with no error — that is what a DDB read of a missing item yields. Any other length is a foreign blob and is an error: the format has no magic and no length prefix, so a 200-byte HLL sketch or a Bloom filter used to decode to a Set=true observation assembled from its first 17 bytes, and that fabricated value then merged into the row and stayed there.
type Option ¶ added in v0.2.0
type Option func(*decayedConfig)
Option configures the decayed-sum monoids. The same options apply to DecayedSum and DecayedSumBytes; WithDecodeErrorHandler is only consulted by the bytes variant, which is the only one that parses a wire form.
func WithDecodeErrorHandler ¶ added in v0.2.0
WithDecodeErrorHandler installs a callback invoked when DecayedSumBytes' Combine is handed bytes that are not a Decayed wire form.
Combine returns no error — the contract is Combine(a, b) V — so the recovery is to keep the operand that decoded and discard the other. Doing that silently is the problem: the wire form is a fixed 17 bytes with no magic and no length prefix, so any longer blob (an HLL sketch, a Bloom filter, a row from a pipeline that changed monoids) used to decode to a Set=true value built out of whatever the first 17 bytes happened to be, and that garbage then merged into the row as if it were real.
The handler must be cheap and non-blocking; it runs on the merge path.
Source Files