From 74282ac72f8c4aa874b1507842a8465cd5d6664f Mon Sep 17 00:00:00 2001 From: Alex Kuznicki Date: Sat, 29 Aug 2026 19:33:30 -0600 Subject: [PATCH 1/2] fixes --- llo/protocol/calculated/calculated.go | 8 - llo/protocol/calculated/decimalmath.go | 7 +- llo/protocol/calculated/doc.go | 3 - llo/protocol/calculated/evaluate_fuzz_test.go | 1 - .../calculated/functions_bench_test.go | 134 ----- llo/protocol/calculated/functions_twap.go | 455 ----------------- .../calculated/functions_twap_test.go | 468 ------------------ llo/protocol/calculated/history_ast.go | 73 --- llo/protocol/calculated/process_fuzz_test.go | 1 - llo/protocol/calculated/program.go | 3 +- llo/protocol/calculated/series.go | 2 +- llo/protocol/calculated/validation_test.go | 62 +-- 12 files changed, 6 insertions(+), 1211 deletions(-) delete mode 100644 llo/protocol/calculated/functions_twap.go delete mode 100644 llo/protocol/calculated/functions_twap_test.go diff --git a/llo/protocol/calculated/calculated.go b/llo/protocol/calculated/calculated.go index bea0c85..0b0122d 100644 --- a/llo/protocol/calculated/calculated.go +++ b/llo/protocol/calculated/calculated.go @@ -78,10 +78,6 @@ var defaultEnv = map[string]any{ "SMA": SMA, "WMA": WMA, "EMA": EMA, - // TWAP needs the round's observation timestamp to anchor its window, so - // NewEnv rebinds it per round. This default only reports that it was called - // against an environment NewEnv did not build. - "TWAP": twapUnbound, // History is rewritten away at compile time (see history_ast.go). It is // registered only so that a call surviving to evaluation fails loudly // instead of resolving to an undefined identifier or, worse, to something @@ -188,10 +184,6 @@ func (e environment) release() { func NewEnv(observationTimestampNanoseconds uint64) environment { env := pool.Get().(environment) env["observations_timestamp"] = observationTimestampNanoseconds - // TWAP's window is anchored on the round's consensus observation timestamp, - // not on the data, so it is bound per round. release() restores the default - // binding, which fails if called. - env["TWAP"] = twapFunc(observationTimestampNanoseconds) return env } diff --git a/llo/protocol/calculated/decimalmath.go b/llo/protocol/calculated/decimalmath.go index 1b0c959..4c348ff 100644 --- a/llo/protocol/calculated/decimalmath.go +++ b/llo/protocol/calculated/decimalmath.go @@ -16,8 +16,6 @@ import ( // racing with a read can yield a corrupted value rather than merely a stale one — // which in a consensus path means two nodes disagreeing, or a panic. // -// TWAP makes it far more likely by calling ln and exp once per bucket. -// // The lock is taken per call rather than per evaluation to keep hold times short. // The cost is negligible against the arithmetic it guards. var transcendentalMu sync.Mutex @@ -173,8 +171,7 @@ func decimalToInt(name string, d decimal.Decimal, minimum, maximum int64) (int, // // 1. No float64. math.Log and math.Exp are not guaranteed bit-identical across // architectures or Go versions, so all logarithms and exponentials go through -// decimal.Ln and decimal.ExpTaylor at a fixed precision. This is why the TWAP -// implementation here is a port of the mercury float-based one, not a reuse. +// decimal.Ln and decimal.ExpTaylor at a fixed precision. // 2. No reliance on decimal.DivisionPrecision. That is a mutable package-level // global: anything in the process can change it and silently move every Div // result. Every division here passes an explicit precision (divRound). @@ -227,7 +224,7 @@ func ln(x decimal.Decimal) (decimal.Decimal, error) { // result, not of the input: exp(1e6) has ~434,000 digits and does not complete in // any useful time. Callers currently only pass logarithms of stored values, which // MaxDecimalExponent already bounds to about ±2302, so the limit is not reachable -// through TWAP today — it is here so that stays true if another caller appears. +// today — it is here so that stays true if a caller appears. func exp(x decimal.Decimal) (decimal.Decimal, error) { if x.Abs().GreaterThan(decimal.NewFromInt(maxExpArgument)) { return decimal.Decimal{}, fmt.Errorf("exponential argument %s exceeds the maximum magnitude of %d", x, maxExpArgument) diff --git a/llo/protocol/calculated/doc.go b/llo/protocol/calculated/doc.go index 13baf24..7648755 100644 --- a/llo/protocol/calculated/doc.go +++ b/llo/protocol/calculated/doc.go @@ -25,8 +25,6 @@ // // Avg(History(s10001, 10)) // EMA(History(s10001, 50), 20) -// TWAP(History(s10001, 600), {window: Duration("5m"), minSamples: 240, -// maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30}) // // The call is both the declaration of how much history to persist and the read of // it. There is no separate configuration: the depth kept for a stream is the @@ -67,7 +65,6 @@ // SMA(w, n) simple mean of the newest n // WMA(w, n) linearly weighted, newest weighted n and the oldest of the n weighted 1 // EMA(w, n) seeded with the mean of the oldest n, then alpha = 2/(n+1) newest-ward -// TWAP(w, c) time-weighted average price over c.window, filling gaps by type // // A window may only be passed directly to one of these. Add(History(s1, 10), 2) is // rejected when the expression is validated, not left to fail during evaluation. diff --git a/llo/protocol/calculated/evaluate_fuzz_test.go b/llo/protocol/calculated/evaluate_fuzz_test.go index 60dcf6c..42bc528 100644 --- a/llo/protocol/calculated/evaluate_fuzz_test.go +++ b/llo/protocol/calculated/evaluate_fuzz_test.go @@ -55,7 +55,6 @@ func FuzzEvaluateExpression(f *testing.F) { "Sum(History(s1, 3))", "Min(History(s1, 3))", "Max(History(s1, 3))", - `TWAP(History(s1, 3), {window: Duration("3s"), minSamples: 1, maxHeadGap: 3, maxInteriorGap: 3, maxTailGap: 3})`, "Ln(s1)", "Log(s1, s2)", "Pow(s1, s2)", diff --git a/llo/protocol/calculated/functions_bench_test.go b/llo/protocol/calculated/functions_bench_test.go index acfa85b..7cc5a02 100644 --- a/llo/protocol/calculated/functions_bench_test.go +++ b/llo/protocol/calculated/functions_bench_test.go @@ -18,11 +18,6 @@ import ( // The figure to keep in mind is the round interval, on the order of a second: // every expression of every channel is evaluated inside one state transition, so // per-expression cost multiplies by the channel count. -// -// TWAP is the one to watch. It takes a logarithm per observed bucket and an -// exponential per bucket in the window, and all of those serialize on the -// process-wide transcendental lock (see decimalmath.go), so its cost does not -// parallelize across the plugin instances sharing a process. func benchSeries(depth int, intervalSeconds int) Series { values := make([]decimal.Decimal, 0, depth) @@ -73,134 +68,6 @@ func BenchmarkWindowFunctions(b *testing.B) { } } -// BenchmarkTWAP measures the settlement-window sizes an operator would actually -// configure. -func BenchmarkTWAP(b *testing.B) { - for _, windowSeconds := range []int{60, 300, 900} { - // One record per second, fully covering the window. - window := benchSeries(windowSeconds, 1) - anchorNs := uint64(windowSeconds+1) * uint64(time.Second) - cfg := map[string]any{ - "window": time.Duration(windowSeconds) * time.Second, - "minSamples": windowSeconds / 2, - "maxHeadGap": windowSeconds, - "maxInteriorGap": windowSeconds, - "maxTailGap": windowSeconds, - } - twap := twapFunc(anchorNs) - - b.Run(fmt.Sprintf("window=%ds", windowSeconds), func(b *testing.B) { - for range b.N { - if _, err := twap(window, cfg); err != nil { - b.Fatal(err) - } - } - }) - } -} - -// BenchmarkTWAPSparse is the worst case for the filling strategy: only interior -// interpolation needs log space, so cost scales with how much of the window is -// missing. A window observed every Nth second is the expensive shape. -func BenchmarkTWAPSparse(b *testing.B) { - const windowSeconds = 300 - - for _, everyNth := range []int{1, 2, 5, 30} { - depth := windowSeconds / everyNth - window := benchSeries(depth, everyNth) - anchorNs := uint64(windowSeconds+1) * uint64(time.Second) - cfg := map[string]any{ - "window": time.Duration(windowSeconds) * time.Second, - "minSamples": 1, - "maxHeadGap": windowSeconds, - "maxInteriorGap": windowSeconds, - "maxTailGap": windowSeconds, - } - twap := twapFunc(anchorNs) - - b.Run(fmt.Sprintf("observedEvery=%ds", everyNth), func(b *testing.B) { - for range b.N { - if _, err := twap(window, cfg); err != nil { - b.Fatal(err) - } - } - }) - } -} - -// BenchmarkTWAPRealisticGaps measures the worst case a production acceptance rule -// actually admits. -// -// The permissive thresholds in BenchmarkTWAPSparse exist to force interpolation -// and show where the cost lives; they are not deployable. With the spec's example -// thresholds (minSamples 240 of 300, maxInteriorGap 10) a window can be missing at -// most 60 buckets, so interpolation is bounded no matter how the gaps fall. -func BenchmarkTWAPRealisticGaps(b *testing.B) { - const windowSeconds = 300 - const minSamples = 240 - - // 240 observations in a 300-second window, with the 60 missing buckets spread - // as 20 interior gaps of 3 — within a maxInteriorGap of 10. - values := make([]decimal.Decimal, 0, minSamples) - timestamps := make([]uint64, 0, minSamples) - second := 0 - for len(values) < minSamples && second < windowSeconds { - if second%15 >= 12 { // 3 missing out of every 15 - second++ - continue - } - values = append(values, decimal.New(int64(110000000000000000+second), -8)) - timestamps = append(timestamps, uint64(second+1)*uint64(time.Second)) - second++ - } - window, err := NewSeries(values, timestamps) - if err != nil { - b.Fatal(err) - } - - cfg := map[string]any{ - "window": time.Duration(windowSeconds) * time.Second, - "minSamples": len(values), - "maxHeadGap": 10, - "maxInteriorGap": 10, - "maxTailGap": 10, - } - twap := twapFunc(uint64(windowSeconds+1) * uint64(time.Second)) - - b.ReportMetric(float64(windowSeconds-len(values)), "missingBuckets") - b.ResetTimer() - for range b.N { - if _, err := twap(window, cfg); err != nil { - b.Fatal(err) - } - } -} - -// BenchmarkTWAPParallel shows what the transcendental lock costs when several -// plugin instances evaluate TWAP at once. Compare ns/op against the serial -// benchmark: no speedup means the lock is the limit. -func BenchmarkTWAPParallel(b *testing.B) { - const windowSeconds = 300 - window := benchSeries(windowSeconds, 1) - anchorNs := uint64(windowSeconds+1) * uint64(time.Second) - cfg := map[string]any{ - "window": time.Duration(windowSeconds) * time.Second, - "minSamples": windowSeconds / 2, - "maxHeadGap": windowSeconds, - "maxInteriorGap": windowSeconds, - "maxTailGap": windowSeconds, - } - twap := twapFunc(anchorNs) - - b.RunParallel(func(pb *testing.PB) { - for pb.Next() { - if _, err := twap(window, cfg); err != nil { - b.Fatal(err) - } - } - }) -} - // BenchmarkProcessCalculatedStreams measures a whole round's expression work, // which is what the round budget actually has to absorb. func BenchmarkProcessCalculatedStreams(b *testing.B) { @@ -212,7 +79,6 @@ func BenchmarkProcessCalculatedStreams(b *testing.B) { {"scalar", "Add(s1, s2)", 0}, {"avg/depth=300", "Avg(History(s1, 300))", 300}, {"ema/depth=300", "EMA(History(s1, 300), 20)", 300}, - {"twap/window=300", `TWAP(History(s1, 300), {window: Duration("5m"), minSamples: 150, maxHeadGap: 300, maxInteriorGap: 300, maxTailGap: 300})`, 300}, } { for _, channels := range []int{1, 32} { b.Run(fmt.Sprintf("%s/channels=%d", tc.name, channels), func(b *testing.B) { diff --git a/llo/protocol/calculated/functions_twap.go b/llo/protocol/calculated/functions_twap.go deleted file mode 100644 index c5686f0..0000000 --- a/llo/protocol/calculated/functions_twap.go +++ /dev/null @@ -1,455 +0,0 @@ -package calculated - -import ( - "errors" - "fmt" - "sort" - "strings" - "time" - - "github.com/shopspring/decimal" -) - -// TWAP is ported from the original spec (ADR 0013/0014/0015), semantics are unchanged. -// It is a port rather than a reuse for two reasons: the source works from decoded -// values and a clock, while here the input is an already-agreed history window; -// and the source computes in float64, which is not guaranteed bit-identical -// across architectures and so cannot appear in a consensus path. Every logarithm, -// exponential and division below is decimal at a fixed precision. -var ( - // ErrTWAPRejected is what every rejection satisfies errors.Is against, so - // callers can detect a rejected window without inspecting the reasons. - ErrTWAPRejected = errors.New("TWAP window rejected by the acceptance rule") - - // ErrTWAPConfig is returned for a malformed configuration. Configuration is - // static, so this is a deployment error rather than a data condition. - ErrTWAPConfig = errors.New("invalid TWAP configuration") -) - -// TWAPRejectionReason enumerates why a window failed the acceptance rule. A -// window can fail several checks at once. -type TWAPRejectionReason string - -const ( - // ReasonInsufficientSamples: M < minSamples, the coverage floor. - ReasonInsufficientSamples TWAPRejectionReason = "min_samples" - // ReasonHeadGapTooLong: Ghead > maxHeadGap, backfilled prefix too long. - ReasonHeadGapTooLong TWAPRejectionReason = "head_gap_too_long" - // ReasonInteriorGapTooLong: Gint > maxInteriorGap, longest both-sides-anchored gap too long. - ReasonInteriorGapTooLong TWAPRejectionReason = "interior_gap_too_long" - // ReasonTailGapTooLong: Gtail > maxTailGap, carry-forward suffix too long. - ReasonTailGapTooLong TWAPRejectionReason = "tail_gap_too_long" -) - -// TWAPRejection carries the measured statistics alongside the thresholds they -// failed, so an operator can tell a thin window from a stalled feed without -// reproducing the calculation. -type TWAPRejection struct { - Reasons []TWAPRejectionReason - M, Ghead, Gint, Gtail int - MinSamples, MaxHeadGap, MaxInteriorGap, MaxTailGap int - WindowStartSeconds, WindowEndSeconds int64 - Records int -} - -func (e *TWAPRejection) Error() string { - reasons := make([]string, 0, len(e.Reasons)) - for _, reason := range e.Reasons { - reasons = append(reasons, string(reason)) - } - return fmt.Sprintf("TWAP: window [%d, %d) rejected (%s): M=%d/%d Ghead=%d/%d Gint=%d/%d Gtail=%d/%d from %d records", - e.WindowStartSeconds, e.WindowEndSeconds, strings.Join(reasons, ","), - e.M, e.MinSamples, e.Ghead, e.MaxHeadGap, e.Gint, e.MaxInteriorGap, e.Gtail, e.MaxTailGap, e.Records) -} - -func (e *TWAPRejection) Is(target error) bool { return target == ErrTWAPRejected } - -// twapConfig is the acceptance rule for one window size. Every field is required: -// a defaulted threshold would silently accept a window an operator never approved. -type twapConfig struct { - windowSeconds int64 - minSamples int - maxHeadGap int - maxInteriorGap int - maxTailGap int -} - -// twapBucket is one second of the dense series the specification operates on. -// price is only meaningful when observed is true. -// -// The specification is written in log-price space throughout: build X[i] = ln(P[i]), -// fill gaps in that space, then average exp of the filled series. This stores the -// price instead, and moves into log space only where filling actually requires it. -// -// That is not a shortcut, it is the same series. For an observed bucket the -// specification computes exp(ln(P)) = P. For a head gap it backfills the first -// observed log-price, and for a tail gap it carries the last, so exponentiating -// those yields that same observed price. Only interior interpolation produces a -// value that is not already a price. -// -// The reason it matters is cost. Filling in log space needs a logarithm per -// observed bucket and an exponential per bucket in the window — about 600 -// operations for a five-minute window — and they all serialize on the -// transcendental lock. Measured: 197ms per evaluation for a 300-second window, -// and 7.2s for 32 such channels in one round, against a round budget on the -// order of a second. Doing it this way, a fully covered window needs no -// transcendental operations at all, and a window with gaps needs two logarithms -// per gap plus one exponential per missing bucket. -type twapBucket struct { - observed bool - price decimal.Decimal -} - -// twapFunc returns the TWAP function bound to a round's consensus observation -// timestamp, which anchors the window. -// -// The anchor has to come from the round rather than from the data: taking it from -// the newest record would silently shorten the window whenever a feed stalled, -// which is exactly the condition the acceptance rule exists to catch. -func twapFunc(observationTimestampNanoseconds uint64) func(any, any) (decimal.Decimal, error) { - return func(x any, rawConfig any) (decimal.Decimal, error) { - series, err := window("TWAP", x) - if err != nil { - return decimal.Decimal{}, err - } - cfg, err := parseTWAPConfig(rawConfig) - if err != nil { - return decimal.Decimal{}, err - } - // NOTE: whether the requested history depth can ever supply minSamples - // observations is a static property of the configuration, and is checked - // at configuration time rather than here. Checking it here would turn a - // specification-defined rejection (M < minSamples, a data condition with - // diagnostics) into a configuration error, losing the measured - // statistics an operator needs. - return twap(series, cfg, int64(observationTimestampNanoseconds/uint64(time.Second))) - } -} - -// twapUnbound is the default TWAP binding. NewEnv replaces it with a function -// bound to the round's observation timestamp; reaching this one means TWAP was -// called against an environment that was not built by NewEnv. -func twapUnbound(any, any) (decimal.Decimal, error) { - return decimal.Decimal{}, errors.New("TWAP has no observation timestamp bound; the environment was not created by NewEnv") -} - -func twap(series Series, cfg twapConfig, anchorSeconds int64) (decimal.Decimal, error) { - windowStart := anchorSeconds - cfg.windowSeconds - buckets := make([]twapBucket, cfg.windowSeconds) - - values, timestamps := series.Values(), series.Timestamps() - for i, ts := range timestamps { - seconds := int64(ts / uint64(time.Second)) - if seconds < windowStart || seconds >= anchorSeconds { - continue // outside the half-open window (ADR 0013) - } - // The price must be positive: the filling rules are defined in log space, - // so a non-positive price has no representation there. Checked here for - // every observed bucket rather than only where a logarithm is taken, so - // acceptance does not depend on where the gaps happen to fall. - if !values[i].IsPositive() { - return decimal.Decimal{}, fmt.Errorf("TWAP: record %d: price %s must be positive", i, values[i]) - } - // Timestamps are strictly increasing, so a later record legitimately - // overwrites an earlier one in the same bucket: newest wins. - buckets[seconds-windowStart] = twapBucket{observed: true, price: values[i]} - } - - m, gHead, gInt, gTail := twapGapStats(buckets) - - var reasons []TWAPRejectionReason - // A floor of 1 observation is required for head backfill to have an anchor. - // With a validated minSamples >= 1 this is redundant, but it keeps a - // misconfiguration from reaching an out-of-range index below. - minSamples := max(cfg.minSamples, 1) - if m < minSamples { - reasons = append(reasons, ReasonInsufficientSamples) - } - if gHead > cfg.maxHeadGap { - reasons = append(reasons, ReasonHeadGapTooLong) - } - if gInt > cfg.maxInteriorGap { - reasons = append(reasons, ReasonInteriorGapTooLong) - } - if gTail > cfg.maxTailGap { - reasons = append(reasons, ReasonTailGapTooLong) - } - if len(reasons) > 0 { - return decimal.Decimal{}, &TWAPRejection{ - Reasons: reasons, - M: m, Ghead: gHead, Gint: gInt, Gtail: gTail, - MinSamples: cfg.minSamples, MaxHeadGap: cfg.maxHeadGap, - MaxInteriorGap: cfg.maxInteriorGap, MaxTailGap: cfg.maxTailGap, - WindowStartSeconds: windowStart, WindowEndSeconds: anchorSeconds, - Records: series.Len(), - } - } - - return twapFillThenAverage(buckets) -} - -// twapGapStats measures M, Ghead, Gint and Gtail by classifying each missing run -// by its position (spec §2, ADR 0015). -// -// Ghead and Gtail are kept separate from Gint deliberately: Gint is the -// both-sides-anchored statistic, and a head or tail run has only one anchor. A -// run spanning the whole window is classified as none of them because it has no -// anchors at all; such a window is always rejected by the M check. -func twapGapStats(buckets []twapBucket) (m, gHead, gInt, gTail int) { - n := len(buckets) - for i := 0; i < n; { - runStart := i - observed := buckets[i].observed - for i < n && buckets[i].observed == observed { - i++ - } - runLen := i - runStart - - if observed { - m += runLen - continue - } - switch { - case runStart == 0 && i == n: - // Entire window missing: no anchors, so not head, tail or interior. - case runStart == 0: - gHead = runLen - case i == n: - gTail = runLen - default: - gInt = max(gInt, runLen) - } - } - return m, gHead, gInt, gTail -} - -// twapFillThenAverage fills every bucket per spec §4 and returns the mean price -// over the full window. -// -// Callers must only reach this once the acceptance rule has passed, which -// guarantees at least one observation. -func twapFillThenAverage(buckets []twapBucket) (decimal.Decimal, error) { - n := len(buckets) - filled := make([]decimal.Decimal, n) - - for i := 0; i < n; { - if buckets[i].observed { - filled[i] = buckets[i].price // spec §4.1: X[i] passes through - i++ - continue - } - runStart := i - for i < n && !buckets[i].observed { - i++ - } - switch { - case runStart == 0: - // Head gap: backfill the first observed price (ADR 0015). - // buckets[i] is observed, because a window with no observation at - // all was rejected above. - for k := 0; k < i; k++ { - filled[k] = buckets[i].price - } - case i == n: - // Tail gap: carry forward the last observed price (spec §4.3). - for k := runStart; k < n; k++ { - filled[k] = buckets[runStart-1].price - } - default: - // Interior gap: log-linear interpolation between the bracketing - // anchors at runStart-1 and i (spec §4.2). This is the only case - // that needs log space, so it is the only one that pays for it. - if err := twapInterpolate(buckets, filled, runStart, i); err != nil { - return decimal.Decimal{}, err - } - } - } - - // TWAP = mean over N (spec §4-5, denominator N not M). - sum := decimal.Zero - for _, price := range filled { - sum = sum.Add(price) - } - return divRoundByInt(sum, n, precision) -} - -// twapInterpolate fills the missing run [runStart, rightIdx) between its -// bracketing anchors (spec §4.2). -// -// Linear interpolation in log space is geometric interpolation in price space: a -// gap between 100 and 1600 fills as 200, 400, 800, not as evenly spaced prices. -// So rather than exponentiating each interpolated log-price, this takes the -// constant per-second ratio once and steps through the gap by multiplication: -// -// ratio = (right / left) ^ (1 / span) -// filled[k] = filled[k-1] * ratio -// -// One power per gap instead of two logarithms plus one exponential per missing -// bucket. With the spec's example thresholds a window can be missing 60 buckets, -// which cost ~73ms the other way and a fraction of that here. Exponentials are the -// expensive operation (~0.5ms each) and reducing their precision only helps by -// about a factor of two, so cutting their number is the only lever that matters. -// -// Determinism: the ratio is computed at a fixed precision and every step is -// rounded, so the sequence is reproducible — the same requirement EMA has, for the -// same reason. -func twapInterpolate(buckets []twapBucket, filled []decimal.Decimal, runStart, rightIdx int) error { - leftIdx := runStart - 1 - left, right := buckets[leftIdx].price, buckets[rightIdx].price - - growth, err := divRound(right, left, doublePrecision) - if err != nil { - return fmt.Errorf("TWAP: bucket %d: %w", leftIdx, err) - } - exponent, err := divRoundByInt(decimal.NewFromInt(1), rightIdx-leftIdx, doublePrecision) - if err != nil { - return err - } - ratio, err := decimalPow(growth, exponent, doublePrecision) - if err != nil { - return fmt.Errorf("TWAP: interpolating buckets %d..%d: %w", runStart, rightIdx-1, err) - } - - price := left - for k := runStart; k < rightIdx; k++ { - price = price.Mul(ratio).Round(doublePrecision) - filled[k] = price - } - return nil -} - -// parseTWAPConfig decodes and validates the configuration map. -// -// Every key is required and no key is optional: a defaulted threshold would mean -// accepting a window against a rule nobody wrote down. Unknown keys are rejected -// too, so a typo fails loudly instead of leaving a threshold at its intended -// value by accident. -func parseTWAPConfig(raw any) (twapConfig, error) { - fields, ok := raw.(map[string]any) - if !ok { - return twapConfig{}, fmt.Errorf("%w: expected a configuration map, got %T", ErrTWAPConfig, raw) - } - - const ( - keyWindow = "window" - keyMinSamples = "minSamples" - keyMaxHeadGap = "maxHeadGap" - keyMaxInteriorGap = "maxInteriorGap" - keyMaxTailGap = "maxTailGap" - ) - known := map[string]bool{ - keyWindow: true, keyMinSamples: true, keyMaxHeadGap: true, - keyMaxInteriorGap: true, keyMaxTailGap: true, - } - unknown := make([]string, 0) - for key := range fields { - if !known[key] { - unknown = append(unknown, key) - } - } - if len(unknown) > 0 { - sort.Strings(unknown) - return twapConfig{}, fmt.Errorf("%w: unknown keys %s", ErrTWAPConfig, strings.Join(unknown, ", ")) - } - - windowSeconds, err := twapWindowSeconds(fields[keyWindow]) - if err != nil { - return twapConfig{}, err - } - minSamples, err := twapConfigInt(fields, keyMinSamples, 1) - if err != nil { - return twapConfig{}, err - } - maxHeadGap, err := twapConfigInt(fields, keyMaxHeadGap, 0) - if err != nil { - return twapConfig{}, err - } - maxInteriorGap, err := twapConfigInt(fields, keyMaxInteriorGap, 0) - if err != nil { - return twapConfig{}, err - } - maxTailGap, err := twapConfigInt(fields, keyMaxTailGap, 0) - if err != nil { - return twapConfig{}, err - } - - if int64(minSamples) > windowSeconds { - return twapConfig{}, fmt.Errorf("%w: minSamples %d exceeds the %d one-second buckets in the window", - ErrTWAPConfig, minSamples, windowSeconds) - } - - return twapConfig{ - windowSeconds: windowSeconds, - minSamples: minSamples, - maxHeadGap: maxHeadGap, - maxInteriorGap: maxInteriorGap, - maxTailGap: maxTailGap, - }, nil -} - -// twapWindowSeconds resolves the window length, which must be a whole number of -// seconds because the calculation is defined over one-second buckets. -func twapWindowSeconds(raw any) (int64, error) { - if raw == nil { - return 0, fmt.Errorf("%w: window is required", ErrTWAPConfig) - } - - var nanoseconds decimal.Decimal - switch v := raw.(type) { - case time.Duration: - nanoseconds = decimal.NewFromInt(int64(v)) - default: - d, err := toDecimal(raw) - if err != nil { - return 0, fmt.Errorf("%w: window: %s", ErrTWAPConfig, err) - } - nanoseconds = d - } - - perSecond := decimal.NewFromInt(int64(time.Second)) - if !nanoseconds.Mod(perSecond).IsZero() { - return 0, fmt.Errorf("%w: window must be a whole number of seconds", ErrTWAPConfig) - } - // DivRound rather than Div: Div reads the mutable decimal.DivisionPrecision - // global. The division is exact here, but the rule holds everywhere. - // Compared and bounded as a decimal, before any narrowing; see decimalToInt. - // The upper bound also caps the per-evaluation work: the calculation - // allocates and fills one bucket per second of the window. - secondsDecimal := nanoseconds.DivRound(perSecond, 0) - if secondsDecimal.LessThan(decimal.NewFromInt(1)) { - return 0, fmt.Errorf("%w: window must be at least one second, got %s", ErrTWAPConfig, secondsDecimal) - } - if secondsDecimal.GreaterThan(decimal.NewFromInt(twapMaxWindowSeconds)) { - return 0, fmt.Errorf("%w: window of %s seconds exceeds the maximum of %d", - ErrTWAPConfig, secondsDecimal, twapMaxWindowSeconds) - } - seconds, err := decimalToInt("window", secondsDecimal, 1, twapMaxWindowSeconds) - if err != nil { - return 0, fmt.Errorf("%w: %s", ErrTWAPConfig, err) - } - return int64(seconds), nil -} - -// twapMaxWindowSeconds bounds the number of one-second buckets a single TWAP -// evaluation may allocate and fill. 24 hours is far beyond any settlement window -// while keeping the per-round work bounded. -const twapMaxWindowSeconds = 24 * 60 * 60 - -func twapConfigInt(fields map[string]any, key string, minimum int) (int, error) { - raw, ok := fields[key] - if !ok || raw == nil { - return 0, fmt.Errorf("%w: %s is required", ErrTWAPConfig, key) - } - d, err := toDecimal(raw) - if err != nil { - return 0, fmt.Errorf("%w: %s: %s", ErrTWAPConfig, key, err) - } - // Bounded as a decimal before narrowing; see decimalToInt. The upper bound - // is the window cap, since every one of these counts seconds or samples - // inside a window that cannot itself be longer than that. - value, err := decimalToInt(key, d, int64(minimum), twapMaxWindowSeconds) - if err != nil { - return 0, fmt.Errorf("%w: %s", ErrTWAPConfig, err) - } - return value, nil -} diff --git a/llo/protocol/calculated/functions_twap_test.go b/llo/protocol/calculated/functions_twap_test.go deleted file mode 100644 index 7b68817..0000000 --- a/llo/protocol/calculated/functions_twap_test.go +++ /dev/null @@ -1,468 +0,0 @@ -package calculated - -import ( - "errors" - "fmt" - "testing" - "time" - - "github.com/shopspring/decimal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// twapWindowStartSeconds is the arbitrary window start the ported cases use. -const twapWindowStartSeconds = 100 - -// twapSeries builds a window from one price per bucket, 0 meaning "no observation -// in that second". Buckets are one second apart starting at twapWindowStartSeconds. -// -// This mirrors the mercury calculator tests, where a report marks exactly one -// bucket observed. -func twapSeries(t *testing.T, prices []int64) Series { - t.Helper() - values := make([]decimal.Decimal, 0, len(prices)) - timestamps := make([]uint64, 0, len(prices)) - for i, price := range prices { - if price == 0 { - continue // missing observation - } - values = append(values, decimal.NewFromInt(price)) - timestamps = append(timestamps, uint64(twapWindowStartSeconds+i)*uint64(time.Second)) - } - s, err := NewSeries(values, timestamps) - require.NoError(t, err) - return s -} - -// twapConfigMap is the configuration as an expression would supply it. -func twapConfigMap(windowSeconds, minSamples, maxHeadGap, maxInteriorGap, maxTailGap int) map[string]any { - return map[string]any{ - "window": time.Duration(windowSeconds) * time.Second, - "minSamples": minSamples, - "maxHeadGap": maxHeadGap, - "maxInteriorGap": maxInteriorGap, - "maxTailGap": maxTailGap, - } -} - -// assertClose compares against a hand-computed value with a tolerance. -// -// Exactness is not available here and that is inherent to the algorithm, not a -// shortcut: the specification fills gaps in log-price space, so every bucket goes -// through exp(ln(price)). At any finite precision that round trip is not the -// identity, so a whole-number expectation cannot be matched bit-for-bit. The -// tolerance is many orders of magnitude tighter than any reporting precision. -func assertClose(t *testing.T, want string, got decimal.Decimal) { - t.Helper() - expected, err := decimal.NewFromString(want) - require.NoError(t, err) - - const tolerance = "0.000000000001" // 1e-12 - limit, err := decimal.NewFromString(tolerance) - require.NoError(t, err) - - diff := got.Sub(expected).Abs() - assert.True(t, diff.LessThanOrEqual(limit), - "expected %s (±%s), got %s (off by %s)", want, tolerance, got, diff) -} - -// TestTWAP_FillThenAverage is the mercury TestCalculate_FillThenAverage suite, -// ported case for case. The expected values are the same, which is the point: the -// port changed the arithmetic from float64 to decimal, not the semantics. -func TestTWAP_FillThenAverage(t *testing.T) { - t.Parallel() - - for _, tc := range []struct { - name string - prices []int64 // one entry per bucket; 0 = missing - minSamples int - maxHead int - maxInterior int - maxTail int - want string - wantReasons []TWAPRejectionReason - }{ - { - name: "no gaps: plain average over the full window", - prices: []int64{100, 200, 300}, - minSamples: 3, - want: "200", - }, - { - name: "interior gap at the threshold: log-linear interpolated", - prices: []int64{100, 0, 0, 0, 1600}, // Gint=3 - minSamples: 2, maxInterior: 3, - // Interpolating in log space doubles each step: 100,200,400,800,1600. - want: "620", - }, - { - name: "interior gap one over the threshold: rejected", - prices: []int64{100, 0, 0, 0, 1600}, - minSamples: 2, maxInterior: 2, - wantReasons: []TWAPRejectionReason{ReasonInteriorGapTooLong}, - }, - { - name: "tail gap at the threshold: carried forward from the last observed price", - prices: []int64{100, 200, 300, 0, 0}, // Gtail=2 - minSamples: 3, maxTail: 2, - want: "240", - }, - { - name: "tail gap one over the threshold: rejected", - prices: []int64{100, 200, 300, 0, 0}, - minSamples: 3, maxTail: 1, - wantReasons: []TWAPRejectionReason{ReasonTailGapTooLong}, - }, - { - name: "insufficient samples: rejected even though every gap is within its threshold", - prices: []int64{100, 0, 300, 0, 500}, // M=3, Gint=1 - minSamples: 4, maxInterior: 5, maxTail: 5, - wantReasons: []TWAPRejectionReason{ReasonInsufficientSamples}, - }, - { - name: "head gap at the threshold: backfilled from the first observed price", - prices: []int64{0, 0, 200, 300, 400}, // Ghead=2 - minSamples: 3, maxHead: 2, - want: "260", - }, - { - name: "head gap one over the threshold: rejected", - prices: []int64{0, 0, 200, 300, 400}, - minSamples: 3, maxHead: 1, - wantReasons: []TWAPRejectionReason{ReasonHeadGapTooLong}, - }, - { - // Gint is the both-sides-anchored statistic, so a head run must not - // count toward it. Ghead=2 while maxInterior is 1: if the head run - // leaked into Gint this would reject instead of producing a value. - name: "head run is not counted toward the interior-gap threshold", - prices: []int64{0, 0, 100, 0, 400}, // Ghead=2, Gint=1 - minSamples: 2, maxHead: 2, maxInterior: 1, - want: "180", - }, - { - name: "head and tail gap in the same window, both at their thresholds", - prices: []int64{0, 0, 100, 100, 0, 0}, // Ghead=2, Gtail=2 - minSamples: 2, maxHead: 2, maxTail: 2, - want: "100", - }, - { - name: "multiple applicable reasons are all returned, not just the first", - prices: []int64{100, 0, 0, 0, 0}, // M=1, Gtail=4 - minSamples: 3, maxInterior: 5, maxTail: 2, - wantReasons: []TWAPRejectionReason{ReasonInsufficientSamples, ReasonTailGapTooLong}, - }, - { - name: "head and tail reasons are reported alongside insufficient samples", - prices: []int64{0, 0, 0, 100, 0}, // M=1, Ghead=3, Gtail=1 - minSamples: 3, maxHead: 2, - wantReasons: []TWAPRejectionReason{ReasonInsufficientSamples, ReasonHeadGapTooLong, ReasonTailGapTooLong}, - }, - } { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - windowSeconds := len(tc.prices) - series := twapSeries(t, tc.prices) - cfg := twapConfigMap(windowSeconds, tc.minSamples, tc.maxHead, tc.maxInterior, tc.maxTail) - - // The anchor is the exclusive end of the window. - anchorNs := uint64(twapWindowStartSeconds+windowSeconds) * uint64(time.Second) - got, err := twapFunc(anchorNs)(series, cfg) - - if tc.wantReasons != nil { - require.Error(t, err) - require.ErrorIs(t, err, ErrTWAPRejected) - var rejection *TWAPRejection - require.ErrorAs(t, err, &rejection) - assert.Equal(t, tc.wantReasons, rejection.Reasons) - return - } - require.NoError(t, err) - assertClose(t, tc.want, got) - }) - } -} - -// TestTWAP_GapStats ports the mercury gap-classification cases. These are the -// statistics the acceptance rule is written in terms of, so they are worth -// pinning independently of the averaging. -func TestTWAP_GapStats(t *testing.T) { - t.Parallel() - - // o marks an observed bucket, x a missing one. - const o, x = true, false - - for _, tc := range []struct { - name string - observed []bool - wantM, wantHead, wantInt, wantTail int - }{ - {"all observed", []bool{o, o, o}, 3, 0, 0, 0}, - {"head gap only", []bool{x, x, o, o, o}, 3, 2, 0, 0}, - {"tail gap only", []bool{o, o, o, x, x}, 3, 0, 0, 2}, - {"single interior gap", []bool{o, x, o}, 2, 0, 1, 0}, - {"head and tail gaps", []bool{x, x, o, x, x}, 1, 2, 0, 2}, - {"head and interior gaps", []bool{x, o, x, x, o}, 2, 1, 2, 0}, - {"interior and tail gaps", []bool{o, x, x, x, x}, 1, 0, 0, 4}, - // No anchors at all, so no run is classified; the M check rejects it. - {"no observations", []bool{x, x, x, x, x}, 0, 0, 0, 0}, - {"single observed bucket", []bool{o}, 1, 0, 0, 0}, - {"single missing bucket", []bool{x}, 0, 0, 0, 0}, - {"alternating", []bool{o, x, o, x, o}, 3, 0, 1, 0}, - {"two interior gaps takes the longest", []bool{o, x, o, x, x, o}, 3, 0, 2, 0}, - {"three interior gaps increasing", []bool{o, x, o, x, x, o, x, x, x, o}, 4, 0, 3, 0}, - {"gaps not in order", []bool{o, x, x, x, o, x, o, x, x, o}, 4, 0, 3, 0}, - {"long head gap", []bool{x, x, x, x, x, o, o}, 2, 5, 0, 0}, - {"long tail gap", []bool{o, o, x, x, x, x, x}, 2, 0, 0, 5}, - } { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - buckets := make([]twapBucket, len(tc.observed)) - for i, observed := range tc.observed { - buckets[i] = twapBucket{observed: observed, price: decimal.NewFromInt(1)} - } - m, head, interior, tail := twapGapStats(buckets) - assert.Equal(t, tc.wantM, m, "M") - assert.Equal(t, tc.wantHead, head, "Ghead") - assert.Equal(t, tc.wantInt, interior, "Gint") - assert.Equal(t, tc.wantTail, tail, "Gtail") - }) - } -} - -// TestTWAP_HalfOpenWindow covers ADR 0013: the anchor second belongs to the next -// window, and observations before the window start are dropped. -func TestTWAP_HalfOpenWindow(t *testing.T) { - t.Parallel() - - const windowSeconds = 3 - anchorSeconds := uint64(twapWindowStartSeconds + windowSeconds) - cfg := twapConfigMap(windowSeconds, 1, windowSeconds, windowSeconds, windowSeconds) - - // An observation exactly at the anchor is excluded, leaving the window empty. - series, err := NewSeries( - []decimal.Decimal{decimal.NewFromInt(100)}, - []uint64{anchorSeconds * uint64(time.Second)}, - ) - require.NoError(t, err) - _, err = twapFunc(anchorSeconds*uint64(time.Second))(series, cfg) - require.ErrorIs(t, err, ErrTWAPRejected, "the anchor second belongs to the next window") - - // One second earlier is inside the window. - series, err = NewSeries( - []decimal.Decimal{decimal.NewFromInt(100)}, - []uint64{(anchorSeconds - 1) * uint64(time.Second)}, - ) - require.NoError(t, err) - got, err := twapFunc(anchorSeconds*uint64(time.Second))(series, cfg) - require.NoError(t, err) - assertClose(t, "100", got) - - // An observation before the window start is dropped, so only the in-window - // one counts. - series, err = NewSeries( - []decimal.Decimal{decimal.NewFromInt(999), decimal.NewFromInt(100)}, - []uint64{(twapWindowStartSeconds - 5) * uint64(time.Second), (anchorSeconds - 1) * uint64(time.Second)}, - ) - require.NoError(t, err) - got, err = twapFunc(anchorSeconds*uint64(time.Second))(series, cfg) - require.NoError(t, err) - assertClose(t, "100", got) -} - -// TestTWAP_Deterministic is the property that makes TWAP usable in a consensus -// path: identical inputs give an identical result, with no float64 anywhere. -func TestTWAP_Deterministic(t *testing.T) { - t.Parallel() - - prices := []int64{1234, 0, 1240, 1250, 0, 0, 1300, 1310} - series := twapSeries(t, prices) - cfg := twapConfigMap(len(prices), 3, 2, 2, 2) - anchorNs := uint64(twapWindowStartSeconds+len(prices)) * uint64(time.Second) - - first, err := twapFunc(anchorNs)(series, cfg) - require.NoError(t, err) - for range 30 { - again, err := twapFunc(anchorNs)(series, cfg) - require.NoError(t, err) - require.True(t, first.Equal(again), "TWAP drifted: %s vs %s", first, again) - } -} - -func TestTWAP_ConfigValidation(t *testing.T) { - t.Parallel() - - series := twapSeries(t, []int64{100, 200, 300}) - anchorNs := uint64(twapWindowStartSeconds+3) * uint64(time.Second) - call := func(cfg any) error { - _, err := twapFunc(anchorNs)(series, cfg) - return err - } - - t.Run("every key is required", func(t *testing.T) { - t.Parallel() - // A defaulted threshold would mean accepting a window against a rule - // nobody wrote down. - for _, missing := range []string{"window", "minSamples", "maxHeadGap", "maxInteriorGap", "maxTailGap"} { - cfg := twapConfigMap(3, 1, 0, 0, 0) - delete(cfg, missing) - err := call(cfg) - require.ErrorIs(t, err, ErrTWAPConfig, "missing %s", missing) - require.ErrorContains(t, err, missing) - } - }) - - t.Run("unknown keys are rejected", func(t *testing.T) { - t.Parallel() - cfg := twapConfigMap(3, 1, 0, 0, 0) - cfg["maxHeadGapp"] = 1 - err := call(cfg) - require.ErrorIs(t, err, ErrTWAPConfig) - require.ErrorContains(t, err, "maxHeadGapp") - }) - - t.Run("window must be whole seconds and positive", func(t *testing.T) { - t.Parallel() - cfg := twapConfigMap(3, 1, 0, 0, 0) - cfg["window"] = 1500 * time.Millisecond - require.ErrorContains(t, call(cfg), "whole number of seconds") - - cfg["window"] = time.Duration(0) - require.ErrorContains(t, call(cfg), "at least one second") - - cfg["window"] = 48 * time.Hour - require.ErrorContains(t, call(cfg), "exceeds the maximum") - }) - - t.Run("oversized configuration values are rejected, not wrapped", func(t *testing.T) { - t.Parallel() - // decimal.IntPart narrows through int64 and returns the low 64 bits of - // an oversized value, so 2^64+1 comes back as 1. Bounding after that - // would silently accept a one-second window or a minSamples of 1. TWAP - // configuration is not required to be literal (see checkTWAP), so these - // values can come from stream data, bounded only by MaxDecimalExponent. - wrapped := decimal.RequireFromString("18446744073709551617") - - cfg := twapConfigMap(3, 1, 0, 0, 0) - // A whole number of seconds, so it clears the modulo check first. - cfg["window"] = wrapped.Mul(decimal.NewFromInt(int64(time.Second))) - require.ErrorContains(t, call(cfg), "exceeds the maximum") - - for _, key := range []string{"minSamples", "maxHeadGap", "maxInteriorGap", "maxTailGap"} { - cfg := twapConfigMap(3, 1, 0, 0, 0) - cfg[key] = wrapped - err := call(cfg) - require.ErrorIs(t, err, ErrTWAPConfig, key) - require.ErrorContains(t, err, "at most", key) - } - }) - - t.Run("thresholds must be whole and non-negative", func(t *testing.T) { - t.Parallel() - cfg := twapConfigMap(3, 1, 0, 0, 0) - cfg["maxHeadGap"] = -1 - require.ErrorContains(t, call(cfg), "at least 0") - - cfg = twapConfigMap(3, 1, 0, 0, 0) - cfg["minSamples"] = 0 - require.ErrorContains(t, call(cfg), "at least 1") - - cfg = twapConfigMap(3, 1, 0, 0, 0) - cfg["maxTailGap"] = 1.5 - require.ErrorContains(t, call(cfg), "whole number") - }) - - t.Run("minSamples cannot exceed the window", func(t *testing.T) { - t.Parallel() - require.ErrorContains(t, call(twapConfigMap(3, 4, 0, 0, 0)), "exceeds the") - }) - - t.Run("a window too thin to satisfy minSamples is a rejection, not a config error", func(t *testing.T) { - t.Parallel() - // Whether the requested depth can ever supply minSamples is a static - // property checked at configuration time. At runtime a thin window is a - // data condition, and must come back with the measured statistics. - shallow := twapSeries(t, []int64{100}) - _, err := twapFunc(anchorNs)(shallow, twapConfigMap(3, 3, 0, 2, 2)) - require.ErrorIs(t, err, ErrTWAPRejected) - var rejection *TWAPRejection - require.ErrorAs(t, err, &rejection) - assert.Equal(t, []TWAPRejectionReason{ReasonInsufficientSamples}, rejection.Reasons) - }) - - t.Run("not a configuration map", func(t *testing.T) { - t.Parallel() - require.ErrorIs(t, call(42), ErrTWAPConfig) - }) - - t.Run("not a window", func(t *testing.T) { - t.Parallel() - _, err := twapFunc(anchorNs)(decimal.NewFromInt(1), twapConfigMap(3, 1, 0, 0, 0)) - require.ErrorContains(t, err, "expects a history window") - }) -} - -// TestTWAP_NonPositivePriceRejected covers the log-space requirement: a -// non-positive price has no logarithm, so the window cannot be filled. -func TestTWAP_NonPositivePriceRejected(t *testing.T) { - t.Parallel() - - series, err := NewSeries( - []decimal.Decimal{decimal.NewFromInt(100), decimal.Zero}, - []uint64{twapWindowStartSeconds * uint64(time.Second), (twapWindowStartSeconds + 1) * uint64(time.Second)}, - ) - require.NoError(t, err) - - anchorNs := uint64(twapWindowStartSeconds+3) * uint64(time.Second) - _, err = twapFunc(anchorNs)(series, twapConfigMap(3, 1, 2, 2, 2)) - require.ErrorContains(t, err, "must be positive") -} - -// TestTWAP_Unbound covers the patch-bypass equivalent for TWAP: without a round -// to anchor the window, it must refuse rather than invent one. -func TestTWAP_Unbound(t *testing.T) { - t.Parallel() - - _, err := twapUnbound(nil, nil) - require.ErrorContains(t, err, "no observation timestamp bound") - - // A pooled environment always carries a bound TWAP; release restores the - // unbound default. - env := NewEnv(uint64(5 * time.Second)) - bound, ok := env["TWAP"].(func(any, any) (decimal.Decimal, error)) - require.True(t, ok, "NewEnv must bind TWAP to the round") - env.release() - - series := twapSeries(t, []int64{100, 200, 300}) - _, err = bound(series, twapConfigMap(3, 1, 0, 0, 0)) - require.Error(t, err, "the round anchor is 5s, so the window is far from these observations") -} - -// TestTWAP_RejectionMessage checks an operator can tell what failed without -// reproducing the calculation. -func TestTWAP_RejectionMessage(t *testing.T) { - t.Parallel() - - prices := []int64{100, 0, 0, 0, 0} - series := twapSeries(t, prices) - anchorNs := uint64(twapWindowStartSeconds+len(prices)) * uint64(time.Second) - - _, err := twapFunc(anchorNs)(series, twapConfigMap(len(prices), 3, 0, 5, 2)) - require.Error(t, err) - - var rejection *TWAPRejection - require.ErrorAs(t, err, &rejection) - assert.Equal(t, 1, rejection.M) - assert.Equal(t, 4, rejection.Gtail) - assert.Equal(t, 3, rejection.MinSamples) - assert.Equal(t, 2, rejection.MaxTailGap) - - message := err.Error() - for _, want := range []string{"min_samples", "tail_gap_too_long", "M=1/3", "Gtail=4/2"} { - assert.Contains(t, message, want) - } - assert.True(t, errors.Is(err, ErrTWAPRejected)) - assert.Contains(t, fmt.Sprint(err), "rejected") -} diff --git a/llo/protocol/calculated/history_ast.go b/llo/protocol/calculated/history_ast.go index 14e584b..dc2dd5e 100644 --- a/llo/protocol/calculated/history_ast.go +++ b/llo/protocol/calculated/history_ast.go @@ -35,10 +35,6 @@ import ( // integer. const HistoryFunctionName = "History" -// twapFunctionName is the DSL name TWAP is registered under, shared with the -// static analysis that validates its configuration. -const twapFunctionName = "TWAP" - // Field selects which part of a stored stream value a window projects. One // stored window serves every field, so History(s1, 10), History(s1_bid, 10) and // History(s1_ask, 10) share a single series in state and differ only here. @@ -147,7 +143,6 @@ var ( "SMA": true, "WMA": true, "EMA": true, - "TWAP": true, } ) @@ -223,9 +218,6 @@ func (p *historyPatcher) Visit(node *ast.Node) { p.approved[arg] = true } } - if callee.Value == twapFunctionName { - p.checkTWAP(n) - } } } } @@ -299,71 +291,6 @@ func (p *historyPatcher) rewrite(node *ast.Node, call *ast.CallNode) { p.refByNode[*node] = ref } -// checkTWAP validates a TWAP call against the depth of the window it reads. -// -// This is the static half of TWAP validation: whether a configuration can ever be -// satisfied is a property of the expression, so it belongs here rather than at -// evaluation time, where the same condition would surface as a per-round -// rejection and look like a data problem instead of a deployment mistake. -// -// Only literal configuration can be checked. A configuration built at runtime is -// left to the runtime validation in functions_twap.go, which is stricter but -// later. -func (p *historyPatcher) checkTWAP(call *ast.CallNode) { - if len(call.Arguments) != 2 { - p.errorf("%s takes exactly 2 arguments (history window, configuration), got %d", twapFunctionName, len(call.Arguments)) - return - } - ref, ok := p.refByNode[call.Arguments[0]] - if !ok { - // Not reading a window at all; the position rule reports that. - return - } - config, ok := call.Arguments[1].(*ast.MapNode) - if !ok { - return // not a literal configuration - } - - minSamples, found := twapConfigLiteral(config, "minSamples") - if !found { - return - } - // Compared as int64: minSamples is a literal and can be any integer the - // parser accepted, so narrowing it to the width of ref.Count would let a - // value above 2^32 wrap into a small one and pass. The runtime validation - // still rejects it, but the diagnostic this check exists to give would be - // lost. - if minSamples < 1 { - p.errorf("%s requires minSamples to be at least 1, got %d", twapFunctionName, minSamples) - return - } - if minSamples > int64(ref.Count) { - p.errorf("%s requires at least %d observations but %s only keeps %d records; increase the history depth or lower minSamples", - twapFunctionName, minSamples, ref, ref.Count) - } -} - -// twapConfigLiteral reads an integer-literal value out of a configuration map -// literal, reporting whether it was present and literal. -func twapConfigLiteral(config *ast.MapNode, key string) (int64, bool) { - for _, pair := range config.Pairs { - kv, ok := pair.(*ast.PairNode) - if !ok { - continue - } - name, ok := kv.Key.(*ast.StringNode) - if !ok || name.Value != key { - continue - } - value, ok := kv.Value.(*ast.IntegerNode) - if !ok { - return 0, false - } - return int64(value.Value), true - } - return 0, false -} - // err reports every problem found, including windows left in a position that // cannot consume them. // diff --git a/llo/protocol/calculated/process_fuzz_test.go b/llo/protocol/calculated/process_fuzz_test.go index 27e8b52..cfc66e2 100644 --- a/llo/protocol/calculated/process_fuzz_test.go +++ b/llo/protocol/calculated/process_fuzz_test.go @@ -73,7 +73,6 @@ func FuzzProcessCalculatedStreams(f *testing.F) { "EMA(History(s1, 3), 2)", "SMA(History(s1, 3), 2)", "Stddev(History(s1, 3))", - `TWAP(History(s1, 3), {window: Duration("3s"), minSamples: 1, maxHeadGap: 3, maxInteriorGap: 3, maxTailGap: 3})`, // Deeper than the reader serves: the whole round takes the warmup gate. "Avg(History(s1, 64))", // Two windows in one expression, so binding order is exercised. diff --git a/llo/protocol/calculated/program.go b/llo/protocol/calculated/program.go index d9d46a0..5202638 100644 --- a/llo/protocol/calculated/program.go +++ b/llo/protocol/calculated/program.go @@ -84,8 +84,7 @@ func AnalyzeExpressionHistory(expression string) ([]HistoryRef, error) { // // It is the check to run before a channel definition reaches consensus: it // parses, rewrites History calls, and applies every static rule (argument -// shapes, depth caps, per-expression fan-out, window positions, reserved names, -// TWAP configuration satisfiability). It does not evaluate, so it needs no +// shapes, depth caps, per-expression fan-out, window positions, reserved names). It does not evaluate, so it needs no // stream values and no persisted state, and it is a pure function of the // expression string. // diff --git a/llo/protocol/calculated/series.go b/llo/protocol/calculated/series.go index e5ac10d..fcd38b7 100644 --- a/llo/protocol/calculated/series.go +++ b/llo/protocol/calculated/series.go @@ -187,7 +187,7 @@ type HistoryReader interface { type syntheticHistoryReader struct { // endNanoseconds is the exclusive upper bound of the synthesized // timestamps, which must be the round's observation timestamp: functions - // that place records into a window relative to it (TWAP) would otherwise see + // that place records into a window relative to it would otherwise see // every record fall outside the window. endNanoseconds uint64 // intervalNanoseconds is the spacing between synthesized records. diff --git a/llo/protocol/calculated/validation_test.go b/llo/protocol/calculated/validation_test.go index 418b980..490ab86 100644 --- a/llo/protocol/calculated/validation_test.go +++ b/llo/protocol/calculated/validation_test.go @@ -22,9 +22,8 @@ func TestValidateExpression(t *testing.T) { "Count(History(s1, 10))", "Avg(History(s1_bid, 300))", "Div(Avg(History(s1, 10)), s2)", - "EMA(History(s1, 50), 20)", - `TWAP(History(s1, 600), {window: Duration("5m"), minSamples: 240, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`, - } { + "EMA(History(s1, 50), 20)", + } { assert.NoError(t, ValidateExpression(expression), "expression %q", expression) } }) @@ -46,49 +45,6 @@ func TestValidateExpression(t *testing.T) { }) } -// TestValidateExpression_TWAPSatisfiability covers the static half of TWAP -// validation: a configuration that can never be satisfied is a deployment -// mistake, and saying so here beats letting every round reject the window and -// look like a data problem. -func TestValidateExpression_TWAPSatisfiability(t *testing.T) { - t.Parallel() - - // 600 records can supply 240 observations. - require.NoError(t, ValidateExpression( - `TWAP(History(s1, 600), {window: Duration("5m"), minSamples: 240, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`)) - - // 100 records can never supply 240. - err := ValidateExpression( - `TWAP(History(s1, 100), {window: Duration("5m"), minSamples: 240, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`) - require.ErrorIs(t, err, ErrHistoryExpression) - assert.Contains(t, err.Error(), "only keeps 100 records") - - // Exactly enough is fine. - require.NoError(t, ValidateExpression( - `TWAP(History(s1, 240), {window: Duration("4m"), minSamples: 240, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`)) - - // A minSamples above the width of the record count must not wrap into a - // small value and pass. 2^32+5 would narrow to 5. - err = ValidateExpression( - `TWAP(History(s1, 100), {window: Duration("5m"), minSamples: 4294967301, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`) - require.ErrorIs(t, err, ErrHistoryExpression) - assert.Contains(t, err.Error(), "only keeps 100 records") - - // A non-positive minSamples is reported as such rather than as a depth - // problem, which would read as "requires at least 0 observations". - err = ValidateExpression( - `TWAP(History(s1, 100), {window: Duration("5m"), minSamples: 0, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`) - require.ErrorIs(t, err, ErrHistoryExpression) - assert.Contains(t, err.Error(), "requires minSamples to be at least 1") - - // A non-literal configuration cannot be checked statically; runtime - // validation still applies. - require.NoError(t, ValidateExpression("TWAP(History(s1, 10), cfg)")) - - // Wrong arity is caught. - require.Error(t, ValidateExpression("TWAP(History(s1, 10))")) -} - func TestValidateChannelExpressions(t *testing.T) { t.Parallel() @@ -169,17 +125,3 @@ func TestValidateChannelExpressions(t *testing.T) { require.ErrorContains(t, err, "expression is empty") assert.Contains(t, err.Error(), "abi index: 1") } - -// TestProcessCalculatedStreamsDryRun_Satisfiability checks the offline path -// rejects the same configurations the static analysis does. -func TestProcessCalculatedStreamsDryRun_Satisfiability(t *testing.T) { - t.Parallel() - - require.NoError(t, ProcessCalculatedStreamsDryRun( - `TWAP(History(s1, 300), {window: Duration("5m"), minSamples: 240, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`)) - - err := ProcessCalculatedStreamsDryRun( - `TWAP(History(s1, 10), {window: Duration("5m"), minSamples: 240, maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})`) - require.Error(t, err) - assert.Contains(t, err.Error(), "only keeps 10 records") -} From 8f71ef6b636ed941475f74d3eee47592b485b93e Mon Sep 17 00:00:00 2001 From: Alex Kuznicki Date: Sat, 29 Aug 2026 19:44:29 -0600 Subject: [PATCH 2/2] restore --- llo/protocol/calculated/calculated.go | 10 +++ llo/protocol/calculated/doc.go | 1 + llo/protocol/calculated/history_ast.go | 85 ++++++++++++++++++ llo/protocol/calculated/validation_test.go | 100 +++++++++++++++++++++ llo/protocol/limits.go | 7 ++ 5 files changed, 203 insertions(+) diff --git a/llo/protocol/calculated/calculated.go b/llo/protocol/calculated/calculated.go index 65094f5..c175930 100644 --- a/llo/protocol/calculated/calculated.go +++ b/llo/protocol/calculated/calculated.go @@ -78,6 +78,9 @@ var defaultEnv = map[string]any{ "SMA": SMA, "WMA": WMA, "EMA": EMA, + // TWAP is a recognized DSL function that accepts a history window and a + // configuration map. The implementation is not provided in this package. + "TWAP": twapStub, // History is rewritten away at compile time (see history_ast.go). It is // registered only so that a call surviving to evaluation fails loudly // instead of resolving to an undefined identifier or, worse, to something @@ -91,6 +94,13 @@ func historyCallReached(...any) (decimal.Decimal, error) { return decimal.Decimal{}, fmt.Errorf("%s was not resolved at compile time; this is a bug in expression compilation", HistoryFunctionName) } +// twapStub is the runtime placeholder for TWAP. The function signature is +// kept so expressions referencing TWAP parse and compile; evaluation returns +// an error. +func twapStub(...any) (decimal.Decimal, error) { + return decimal.Decimal{}, errors.New("TWAP is not implemented") +} + var ( pool = sync.Pool{ New: func() any { diff --git a/llo/protocol/calculated/doc.go b/llo/protocol/calculated/doc.go index 7648755..695cc44 100644 --- a/llo/protocol/calculated/doc.go +++ b/llo/protocol/calculated/doc.go @@ -65,6 +65,7 @@ // SMA(w, n) simple mean of the newest n // WMA(w, n) linearly weighted, newest weighted n and the oldest of the n weighted 1 // EMA(w, n) seeded with the mean of the oldest n, then alpha = 2/(n+1) newest-ward +// TWAP(w, c) time-weighted average price (not implemented; recognized but not evaluated) // // A window may only be passed directly to one of these. Add(History(s1, 10), 2) is // rejected when the expression is validated, not left to fail during evaluation. diff --git a/llo/protocol/calculated/history_ast.go b/llo/protocol/calculated/history_ast.go index dc2dd5e..40316d6 100644 --- a/llo/protocol/calculated/history_ast.go +++ b/llo/protocol/calculated/history_ast.go @@ -35,6 +35,15 @@ import ( // integer. const HistoryFunctionName = "History" +// twapFunctionName is the DSL name TWAP is registered under, shared with the +// static analysis that validates its configuration. +const twapFunctionName = "TWAP" + +// twapMaxWindowSeconds bounds the number of one-second buckets a single TWAP +// evaluation may allocate and fill. 24 hours is far beyond any settlement window +// while keeping the per-round work bounded. +const twapMaxWindowSeconds = 24 * 60 * 60 + // Field selects which part of a stored stream value a window projects. One // stored window serves every field, so History(s1, 10), History(s1_bid, 10) and // History(s1_ask, 10) share a single series in state and differ only here. @@ -143,6 +152,7 @@ var ( "SMA": true, "WMA": true, "EMA": true, + "TWAP": true, } ) @@ -175,6 +185,11 @@ type historyPatcher struct { refByNode map[ast.Node]HistoryRef fanOut uint64 + + // twapCalls counts the expression's TWAP calls. Each one can request a + // maximum-length window, so their count is what bounds the bucket work an + // expression can ask for every round. + twapCalls int } func newHistoryPatcher() *historyPatcher { @@ -218,6 +233,9 @@ func (p *historyPatcher) Visit(node *ast.Node) { p.approved[arg] = true } } + if callee.Value == twapFunctionName { + p.checkTWAP(n) + } } } } @@ -291,6 +309,73 @@ func (p *historyPatcher) rewrite(node *ast.Node, call *ast.CallNode) { p.refByNode[*node] = ref } +// checkTWAP validates a TWAP call at compile time: arity, per-expression call +// count, and static satisfiability of minSamples against the history depth. +// +// Only literal configuration can be checked. A configuration built at runtime +// is left to runtime validation, which is stricter but later. +func (p *historyPatcher) checkTWAP(call *ast.CallNode) { + // Counted first, and counted whatever the call looks like: this is the one + // TWAP check that does not depend on the configuration being literal, which + // is what makes it a bound rather than a diagnostic. + p.twapCalls++ + if p.twapCalls > protocol.MaxTWAPCallsPerExpression { + p.errorf("expression makes more than %d %s calls; each may request a window of up to %d one-second buckets, so their number is capped", + protocol.MaxTWAPCallsPerExpression, twapFunctionName, twapMaxWindowSeconds) + return + } + if len(call.Arguments) != 2 { + p.errorf("%s takes exactly 2 arguments (history window, configuration), got %d", twapFunctionName, len(call.Arguments)) + return + } + ref, ok := p.refByNode[call.Arguments[0]] + if !ok { + // Not reading a window at all; the position rule reports that. + return + } + config, ok := call.Arguments[1].(*ast.MapNode) + if !ok { + return // not a literal configuration + } + + minSamples, found := twapConfigLiteral(config, "minSamples") + if !found { + return + } + // Compared as int64: minSamples is a literal and can be any integer the + // parser accepted, so narrowing it to the width of ref.Count would let a + // value above 2^32 wrap into a small one and pass. + if minSamples < 1 { + p.errorf("%s requires minSamples to be at least 1, got %d", twapFunctionName, minSamples) + return + } + if minSamples > int64(ref.Count) { + p.errorf("%s requires at least %d observations but %s only keeps %d records; increase the history depth or lower minSamples", + twapFunctionName, minSamples, ref, ref.Count) + } +} + +// twapConfigLiteral reads an integer-literal value out of a configuration map +// literal, reporting whether it was present and literal. +func twapConfigLiteral(config *ast.MapNode, key string) (int64, bool) { + for _, pair := range config.Pairs { + kv, ok := pair.(*ast.PairNode) + if !ok { + continue + } + name, ok := kv.Key.(*ast.StringNode) + if !ok || name.Value != key { + continue + } + value, ok := kv.Value.(*ast.IntegerNode) + if !ok { + return 0, false + } + return int64(value.Value), true + } + return 0, false +} + // err reports every problem found, including windows left in a position that // cannot consume them. // diff --git a/llo/protocol/calculated/validation_test.go b/llo/protocol/calculated/validation_test.go index 490ab86..8ea051f 100644 --- a/llo/protocol/calculated/validation_test.go +++ b/llo/protocol/calculated/validation_test.go @@ -23,6 +23,7 @@ func TestValidateExpression(t *testing.T) { "Avg(History(s1_bid, 300))", "Div(Avg(History(s1, 10)), s2)", "EMA(History(s1, 50), 20)", + `TWAP(History(s1, 600), {window: Duration("5m"), minSamples: 240})`, } { assert.NoError(t, ValidateExpression(expression), "expression %q", expression) } @@ -125,3 +126,102 @@ func TestValidateChannelExpressions(t *testing.T) { require.ErrorContains(t, err, "expression is empty") assert.Contains(t, err.Error(), "abi index: 1") } + +// TestValidateExpression_TWAPSatisfiability covers the static half of TWAP +// validation: a configuration that can never be satisfied is a deployment +// mistake, and saying so here beats letting every round reject the window and +// look like a data problem. +func TestValidateExpression_TWAPSatisfiability(t *testing.T) { + t.Parallel() + + // 600 records can supply 240 observations. + require.NoError(t, ValidateExpression( + `TWAP(History(s1, 600), {window: Duration("5m"), minSamples: 240})`)) + + // 100 records can never supply 240. + err := ValidateExpression( + `TWAP(History(s1, 100), {window: Duration("5m"), minSamples: 240})`) + require.ErrorIs(t, err, ErrHistoryExpression) + assert.Contains(t, err.Error(), "only keeps 100 records") + + // Exactly enough is fine. + require.NoError(t, ValidateExpression( + `TWAP(History(s1, 240), {window: Duration("4m"), minSamples: 240})`)) + + // A minSamples above the width of the record count must not wrap into a + // small value and pass. 2^32+5 would narrow to 5. + err = ValidateExpression( + `TWAP(History(s1, 100), {window: Duration("5m"), minSamples: 4294967301})`) + require.ErrorIs(t, err, ErrHistoryExpression) + assert.Contains(t, err.Error(), "only keeps 100 records") + + // A non-positive minSamples is reported as such rather than as a depth + // problem, which would read as "requires at least 0 observations". + err = ValidateExpression( + `TWAP(History(s1, 100), {window: Duration("5m"), minSamples: 0})`) + require.ErrorIs(t, err, ErrHistoryExpression) + assert.Contains(t, err.Error(), "requires minSamples to be at least 1") + + // A non-literal configuration cannot be checked statically; runtime + // validation still applies. + require.NoError(t, ValidateExpression("TWAP(History(s1, 10), cfg)")) + + // Wrong arity is caught. + require.Error(t, ValidateExpression("TWAP(History(s1, 10))")) +} + +// TestValidateExpression_TWAPCallCount covers the bucket bound: each TWAP call +// may request a maximum-length window, so the number of them is what limits the +// work one expression can ask for every round. +func TestValidateExpression_TWAPCallCount(t *testing.T) { + t.Parallel() + + require.NoError(t, ValidateExpression(twapCalls(protocol.MaxTWAPCallsPerExpression))) + + err := ValidateExpression(twapCalls(protocol.MaxTWAPCallsPerExpression + 1)) + require.ErrorContains(t, err, "calls") + + // The count is syntactic, so a configuration the other TWAP checks cannot + // read is still counted. This is the case a window-based budget would miss. + opaque := make([]string, 0, protocol.MaxTWAPCallsPerExpression+1) + for i := range protocol.MaxTWAPCallsPerExpression + 1 { + opaque = append(opaque, fmt.Sprintf("TWAP(History(s%d, 1), cfg)", i+1)) + } + require.NoError(t, ValidateExpression(sumExpressions(opaque[:protocol.MaxTWAPCallsPerExpression]))) + require.ErrorContains(t, ValidateExpression(sumExpressions(opaque)), "calls") +} + +// twapCalls builds an expression making count TWAP calls. Each reads its own +// depth-1 window, so the expression stays far inside the history budget however +// many calls it makes — which is the point: history depth does not bound TWAP +// bucket work. +func twapCalls(count int) string { + calls := make([]string, 0, count) + for i := range count { + calls = append(calls, fmt.Sprintf( + `TWAP(History(s%d, 1), {window: Duration("1s"), minSamples: 1})`, + i+1)) + } + return sumExpressions(calls) +} + +func sumExpressions(expressions []string) string { + summed := expressions[0] + for _, expression := range expressions[1:] { + summed = fmt.Sprintf("Add(%s, %s)", summed, expression) + } + return summed +} + +// TestProcessCalculatedStreamsDryRun_Satisfiability checks the offline path +// rejects the same configurations the static analysis does. The satisfiable +// case is not tested here because the TWAP stub returns an error at evaluation; +// only the static rejection path is exercised. +func TestProcessCalculatedStreamsDryRun_Satisfiability(t *testing.T) { + t.Parallel() + + err := ProcessCalculatedStreamsDryRun( + `TWAP(History(s1, 10), {window: Duration("5m"), minSamples: 240})`) + require.Error(t, err) + assert.Contains(t, err.Error(), "only keeps 10 records") +} diff --git a/llo/protocol/limits.go b/llo/protocol/limits.go index 4f12e8b..3fdf878 100644 --- a/llo/protocol/limits.go +++ b/llo/protocol/limits.go @@ -82,6 +82,13 @@ const ( // arbitrarily expensive evaluation. MaxHistoryRecordsPerExpression = 4 * MaxHistoryRecordsPerPair + // MaxTWAPCallsPerExpression bounds how many TWAP calls a single expression + // may make. Each call may request a maximum-length window of one-second + // buckets allocated and filled every round, so the count is what bounds + // the work. Consensus-relevant: every oracle must reject the same + // expression, so it is never per-node configurable. + MaxTWAPCallsPerExpression = 4 + // MaxHistoryRecordBytes is the maximum serialized size of one history // record, enforced on append (StreamHistory.Append) and used as the // per-record size when admitting pairs against MaxHistoryTotalBytes.