diff --git a/generate_golden b/generate_golden new file mode 100755 index 0000000..ad44dd7 Binary files /dev/null and b/generate_golden differ diff --git a/llo/dev/v31/blobpump_test.go b/llo/dev/v31/blobpump_test.go index e0830b4..07ca7da 100644 --- a/llo/dev/v31/blobpump_test.go +++ b/llo/dev/v31/blobpump_test.go @@ -222,8 +222,8 @@ func Test_observableStreams(t *testing.T) { 2: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, 3: {ReportFormat: llotypes.ReportFormatJSON, Tombstone: true, Streams: []llotypes.Stream{{StreamID: 102, Aggregator: llotypes.AggregatorMedian}}}, }} - require.ElementsMatch(t, []llotypes.StreamID{100}, observableStreams(state)) - require.Empty(t, observableStreams(&kvState{})) + require.ElementsMatch(t, []llotypes.StreamID{100}, observableStreams(state, 0, 0)) + require.Empty(t, observableStreams(&kvState{}, 0, 0)) } // Test_blobPump_DisabledIsInert covers hosts that run the plugin without blob diff --git a/llo/dev/v31/doc.go b/llo/dev/v31/doc.go index 19105a3..3456af6 100644 --- a/llo/dev/v31/doc.go +++ b/llo/dev/v31/doc.go @@ -27,8 +27,9 @@ // how many channels and streams exist: // // - r/agg holds the per-round ("hot") state — observation timestamp, -// validAfter watermarks, per-channel reportability, and carry-forward -// timestamped aggregates — and is rewritten every round. +// validAfter watermarks, per-channel reportability, the observation +// schedule, and carry-forward timestamped aggregates — and is rewritten +// every round. // - c/defs holds every channel definition and is rewritten only when the // definitions change; c/seqnr records the sequence number of that write. // - c/lifecycle holds the lifecycle stage and is written only on change. diff --git a/llo/dev/v31/factory.go b/llo/dev/v31/factory.go index 09dd12a..3c92081 100644 --- a/llo/dev/v31/factory.go +++ b/llo/dev/v31/factory.go @@ -94,23 +94,24 @@ func (f *PluginFactory) NewReportingPlugin(ctx context.Context, cfg ocr3types.Re } p := &Plugin{ - Config: f.Config, - PredecessorConfigDigest: onchainConfig.PredecessorConfigDigest, - ConfigDigest: cfg.ConfigDigest, - PredecessorRetirementReportCache: f.PredecessorRetirementReportCache, - ShouldRetireCache: f.ShouldRetireCache, - ChannelDefinitionCache: f.ChannelDefinitionCache, - DataSource: f.DataSource, - Logger: l, - N: cfg.N, - F: cfg.F, - RetirementReportCodec: f.RetirementReportCodec, - ReportCodecs: f.ReportCodecs, - DonID: f.DonID, - OutcomeTelemetryCh: f.OutcomeTelemetryCh, - ReportTelemetryCh: f.ReportTelemetryCh, - ProtocolVersion: offchainConfig.ProtocolVersion, - DefaultMinReportIntervalNanoseconds: offchainConfig.DefaultMinReportIntervalNanoseconds, + Config: f.Config, + PredecessorConfigDigest: onchainConfig.PredecessorConfigDigest, + ConfigDigest: cfg.ConfigDigest, + PredecessorRetirementReportCache: f.PredecessorRetirementReportCache, + ShouldRetireCache: f.ShouldRetireCache, + ChannelDefinitionCache: f.ChannelDefinitionCache, + DataSource: f.DataSource, + Logger: l, + N: cfg.N, + F: cfg.F, + RetirementReportCodec: f.RetirementReportCodec, + ReportCodecs: f.ReportCodecs, + DonID: f.DonID, + OutcomeTelemetryCh: f.OutcomeTelemetryCh, + ReportTelemetryCh: f.ReportTelemetryCh, + ProtocolVersion: offchainConfig.ProtocolVersion, + DefaultMinReportIntervalNanoseconds: offchainConfig.DefaultMinReportIntervalNanoseconds, + DefaultMinObservationIntervalNanoseconds: offchainConfig.DefaultMinObservationIntervalNanoseconds, } // Definitions and the opts decoded from them are cached together, as one diff --git a/llo/dev/v31/flow_test.go b/llo/dev/v31/flow_test.go index f72ca03..b624418 100644 --- a/llo/dev/v31/flow_test.go +++ b/llo/dev/v31/flow_test.go @@ -270,6 +270,90 @@ func Test_StateTransition_ChannelRemoval(t *testing.T) { require.NotContains(t, hot.reportedLastRound, llotypes.ChannelID(1)) } +// Promotion replaces validAfter wholesale from the predecessor's retirement +// report so the handover is gapless. The observation schedule must be cleared to +// match: a schedule carried over from staging can leave a channel not due on the +// promotion round, and a channel that is not aggregated has no values and so +// cannot report, reopening the gap promotion exists to avoid. +func Test_StateTransition_PromotionClearsObservationSchedule(t *testing.T) { + ctx := tests.Context(t) + p := testPlugin(t) + p.DefaultMinReportIntervalNanoseconds = 5000 + p.DefaultMinObservationIntervalNanoseconds = 5000 + predecessor := ocrtypes.ConfigDigest{0xAB} + p.PredecessorConfigDigest = &predecessor + p.PredecessorRetirementReportCache = &mockPredecessorRetirementReportCache{ + // Old watermark: the channel is immediately report-due on promotion. + report: protocol.RetirementReport{ValidAfterNanoseconds: map[llotypes.ChannelID]uint64{1: 1}}, + } + channelDef := llotypes.ChannelDefinition{ + ReportFormat: llotypes.ReportFormatJSON, + Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}, + } + kv := newMemKV() + + _, err := p.StateTransition(ctx, 1, ocrtypes.AttributedQuery{}, []ocrtypes.AttributedObservation{ao(0, nil), ao(1, nil), ao(2, nil)}, kv, testBlobs) + require.NoError(t, err) + _, err = p.StateTransition(ctx, 2, ocrtypes.AttributedQuery{}, addChannelRound(t, 1_000, 1, channelDef), kv, testBlobs) + require.NoError(t, err) + + valued := func(ts uint64, v int64) []ocrtypes.AttributedObservation { + obs := Observation{ + UnixTimestampNanoseconds: ts, + StreamValues: protocol.StreamValues{100: protocol.ToDecimal(decimal.NewFromInt(v))}, + } + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + aos = append(aos, ao(i, mustEncodeObs(t, obs))) + } + return aos + } + + // Run the staging instance until it has reported and so acquired a schedule + // slot some way in the future. + seqNr, ts := uint64(3), uint64(3_000) + for range 6 { + _, err = p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valued(ts, 10), kv, testBlobs) + require.NoError(t, err) + seqNr++ + ts += 3_000 + if storedObservationDue(t, kv, 1) != 0 { + break + } + } + scheduled := storedObservationDue(t, kv, 1) + require.NotZero(t, scheduled, "staging must acquire a schedule slot for this test to mean anything") + + // Promote while that slot is still in the future, which is the case the + // carried-over schedule would break. + ts = scheduled - 1_000 + require.Greater(t, scheduled, ts) + + // Promote. validAfter is reseeded from the retirement report, so the channel + // is report-due immediately; it must also be aggregated immediately. + promo := Observation{UnixTimestampNanoseconds: ts, AttestedPredecessorRetirement: []byte("attested")} + promoAOs := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + promoAOs = append(promoAOs, ao(i, mustEncodeObs(t, promo))) + } + _, err = p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, promoAOs, kv, testBlobs) + require.NoError(t, err) + seqNr++ + ts += 3_000 + + require.Equal(t, string(protocol.LifeCycleStageProduction), string(kv.m[string(keyLifecycle)])) + require.Zero(t, storedObservationDue(t, kv, 1), + "promotion must clear the schedule so the channel is due immediately") + + // The next round must aggregate it, which is what lets it report. + prec, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valued(ts, 20), kv, testBlobs) + require.NoError(t, err) + decoded, err := decodePrecursor(prec) + require.NoError(t, err) + require.NotNil(t, decoded.StreamAggregates[100][llotypes.AggregatorMedian], + "a promoted channel must be aggregated rather than waiting for a staging schedule slot") +} + func Test_StateTransition_Promotion(t *testing.T) { ctx := tests.Context(t) p := testPlugin(t) diff --git a/llo/dev/v31/history_flow_test.go b/llo/dev/v31/history_flow_test.go index ad19ea8..392009f 100644 --- a/llo/dev/v31/history_flow_test.go +++ b/llo/dev/v31/history_flow_test.go @@ -103,6 +103,86 @@ func Test_History_Warmup(t *testing.T) { } } +// Test_History_SurvivesObservationSkip covers the interaction between the +// observation interval and stream history. A channel reading history is not +// exempt from the skip, so its window is sampled at its report cadence: it must +// still gain exactly one record per cycle it is aggregated, gain none on the +// rounds it is skipped, and stay readable across the skip. +func Test_History_SurvivesObservationSkip(t *testing.T) { + ctx := tests.Context(t) + const depth = 3 + const interval = 50_000 + expression := fmt.Sprintf("Count(History(s100, %d))", depth) + + p := historyPlugin(t, expression) + p.DefaultMinReportIntervalNanoseconds = interval + p.DefaultMinObservationIntervalNanoseconds = interval + kv := newMemKV() + bootstrapHistoryChannel(t, p, kv, expression) + + key := histKey{streamID: 100, aggregator: llotypes.AggregatorMedian} + seqNr := uint64(3) + ts := uint64(10_000) + + // Warm up and run to the first report. The channel has never reported, so it + // has no schedule slot and is aggregated - and appends - every round. + var reportedAt uint64 + lastLen := 0 + for range 20 { + _, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, ts, 1), kv, testBlobs) + require.NoError(t, err) + seqNr++ + + grown := readHistory(t, kv, key.streamID, key.aggregator).Len() + require.Greater(t, grown, lastLen, "an unscheduled channel appends every round") + lastLen = grown + + if reportedFlag(t, kv, 1) { + reportedAt = ts + break + } + ts += 10_000 + } + require.NotZero(t, reportedAt, "the channel must report before a skip window can be exercised") + require.Zero(t, storedObservationDue(t, kv, 1), "not scheduled until it has reported") + + // The next round picks up that report and schedules the channel forward. + ts += 10_000 + _, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, ts, 99), kv, testBlobs) + require.NoError(t, err) + seqNr++ + dueAt := storedObservationDue(t, kv, 1) + require.Equal(t, reportedAt+interval, dueAt, "scheduled one interval on from the round that reported") + + // Rounds inside the skip window append nothing. + depthAtSkipStart := readHistory(t, kv, key.streamID, key.aggregator).Len() + skipRounds := 0 + for ts+10_000 < dueAt { + ts += 10_000 + skipRounds++ + _, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, ts, 1), kv, testBlobs) + require.NoError(t, err) + seqNr++ + require.Equal(t, depthAtSkipStart, readHistory(t, kv, key.streamID, key.aggregator).Len(), + "a skipped round must not append to the window") + require.False(t, reportedFlag(t, kv, 1), "a skipped channel does not report") + } + require.NotZero(t, skipRounds, "the test must actually exercise skipped rounds") + + // The window is still there and still at its required depth: the skip lowers + // the sampling rate, it does not tear the window down. + stored := readHistory(t, kv, key.streamID, key.aggregator) + require.NotNil(t, stored) + require.GreaterOrEqual(t, stored.Len(), depth, "the window stays readable across a skip window") + require.Equal(t, uint32(depth), stored.RequiredCount()) + + // Coming due again appends exactly one more record. + _, err = p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, dueAt+10_000, 7), kv, testBlobs) + require.NoError(t, err) + require.Equal(t, depthAtSkipStart+1, readHistory(t, kv, key.streamID, key.aggregator).Len(), + "the round the channel is due again appends exactly one record") +} + // Test_History_EvictsAtDepth checks the window stays bounded across many rounds // and keeps the newest values. func Test_History_EvictsAtDepth(t *testing.T) { @@ -277,9 +357,10 @@ func storedChannelDefinitions(t *testing.T, kv *memKV) llotypes.ChannelDefinitio func storedHotState(t *testing.T, kv *memKV) *kvState { t.Helper() s := &kvState{ - validAfterNanoseconds: map[llotypes.ChannelID]uint64{}, - reportedLastRound: map[llotypes.ChannelID]bool{}, - carryForward: map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{}, + validAfterNanoseconds: map[llotypes.ChannelID]uint64{}, + reportedLastRound: map[llotypes.ChannelID]bool{}, + observationDueNanoseconds: map[llotypes.ChannelID]uint64{}, + carryForward: map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{}, } require.NoError(t, readHotState(kv, s)) return s @@ -291,6 +372,13 @@ func storedValidAfter(t *testing.T, kv *memKV, cid llotypes.ChannelID) uint64 { return storedHotState(t, kv).validAfterNanoseconds[cid] } +// storedObservationDue returns the persisted observation schedule slot, or 0 if +// the channel has none (which means it is due). +func storedObservationDue(t *testing.T, kv *memKV, cid llotypes.ChannelID) uint64 { + t.Helper() + return storedHotState(t, kv).observationDueNanoseconds[cid] +} + // reportedFlag returns the reportability decision the last round persisted. func reportedFlag(t *testing.T, kv *memKV, cid llotypes.ChannelID) bool { t.Helper() diff --git a/llo/dev/v31/kv.go b/llo/dev/v31/kv.go index d4f4fa4..391913f 100644 --- a/llo/dev/v31/kv.go +++ b/llo/dev/v31/kv.go @@ -115,6 +115,11 @@ type kvState struct { // round can advance validAfter faithfully without re-deriving it from // aggregates that are not persisted. reportedLastRound map[llotypes.ChannelID]bool + // observationDueNanoseconds is the observation schedule: when each channel + // next becomes due for observation and aggregation. A channel with no entry + // is due. Distinct from validAfterNanoseconds, which is a report boundary; + // see nextObservationDue for why the two must not be conflated. + observationDueNanoseconds map[llotypes.ChannelID]uint64 // carryForward holds the timestamped aggregates that survive across rounds // (newer-wins monotonicity). Regular aggregates are recomputed fresh every // round and are never persisted. @@ -148,10 +153,11 @@ func loadKVState(r ocr3_1types.KeyValueStateReader, cache *protocol.ChannelCache // StateTransition needs the hot state and must use loadKVState. func loadColdKVState(r ocr3_1types.KeyValueStateReader, cache *protocol.ChannelCache) (*kvState, error) { s := &kvState{ - channelDefinitions: llotypes.ChannelDefinitions{}, - validAfterNanoseconds: map[llotypes.ChannelID]uint64{}, - reportedLastRound: map[llotypes.ChannelID]bool{}, - carryForward: map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{}, + channelDefinitions: llotypes.ChannelDefinitions{}, + validAfterNanoseconds: map[llotypes.ChannelID]uint64{}, + reportedLastRound: map[llotypes.ChannelID]bool{}, + observationDueNanoseconds: map[llotypes.ChannelID]uint64{}, + carryForward: map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{}, } lc, err := r.Read(keyLifecycle) @@ -223,6 +229,9 @@ func readHotState(r ocr3_1types.KeyValueStateReader, s *kvState) error { for _, cid := range pb.ReportableChannelIDs { s.reportedLastRound[cid] = true } + for _, d := range pb.ObservationDueNanoseconds { + s.observationDueNanoseconds[d.ChannelID] = d.DueAtNanoseconds + } for _, sa := range pb.StreamAggregates { sv, err := protocol.UnmarshalProtoStreamValue(sa.StreamValue) if err != nil { @@ -242,6 +251,36 @@ func readHotState(r ocr3_1types.KeyValueStateReader, s *kvState) error { return nil } +// readHotStateForObservation reads the r/agg record and extracts everything the +// Observation phase needs to decide which channels are due - the observation +// schedule, the previous observation timestamp, the validAfter watermarks and +// the reportability flags - while skipping the (potentially large) +// carry-forward stream aggregates, which only StateTransition uses. +func readHotStateForObservation(r ocr3_1types.KeyValueStateReader, s *kvState) error { + b, err := r.Read(keyHotState) + if err != nil { + return fmt.Errorf("read hot state: %w", err) + } + if len(b) == 0 { + return nil + } + pb := &protocol.LLOHotStateProto{} + if err := proto.Unmarshal(b, pb); err != nil { + return fmt.Errorf("unmarshal hot state: %w", err) + } + s.observationTimestampNs = pb.ObservationTimestampNanoseconds + for _, va := range pb.ValidAfterNanoseconds { + s.validAfterNanoseconds[va.ChannelID] = va.ValidAfterNanoseconds + } + for _, cid := range pb.ReportableChannelIDs { + s.reportedLastRound[cid] = true + } + for _, d := range pb.ObservationDueNanoseconds { + s.observationDueNanoseconds[d.ChannelID] = d.DueAtNanoseconds + } + return nil +} + // writeLifecycle persists the lifecycle stage. func writeLifecycle(w ocr3_1types.KeyValueStateReadWriter, stage llotypes.LifeCycleStage) error { return w.Write(keyLifecycle, []byte(stage)) @@ -281,6 +320,7 @@ func writeHotState( observationTimestampNs uint64, validAfterNanoseconds map[llotypes.ChannelID]uint64, reportable map[llotypes.ChannelID]bool, + observationDueNanoseconds map[llotypes.ChannelID]uint64, carryForward map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue, ) error { pb := &protocol.LLOHotStateProto{ @@ -307,6 +347,17 @@ func writeHotState( return pb.ReportableChannelIDs[i] < pb.ReportableChannelIDs[j] }) + pb.ObservationDueNanoseconds = make([]*protocol.LLOChannelIDAndObservationDueProto, 0, len(observationDueNanoseconds)) + for id, dueAt := range observationDueNanoseconds { + pb.ObservationDueNanoseconds = append(pb.ObservationDueNanoseconds, &protocol.LLOChannelIDAndObservationDueProto{ + ChannelID: id, + DueAtNanoseconds: dueAt, + }) + } + sort.Slice(pb.ObservationDueNanoseconds, func(i, j int) bool { + return pb.ObservationDueNanoseconds[i].ChannelID < pb.ObservationDueNanoseconds[j].ChannelID + }) + for sid, aggregates := range carryForward { for agg, tsv := range aggregates { if tsv == nil { diff --git a/llo/dev/v31/kv_test.go b/llo/dev/v31/kv_test.go index 2c6f960..74a6921 100644 --- a/llo/dev/v31/kv_test.go +++ b/llo/dev/v31/kv_test.go @@ -58,7 +58,7 @@ func Test_ChannelCache_StaleSeqNrForcesReload(t *testing.T) { kv := newMemKV() defs := llotypes.ChannelDefinitions{1: jsonChannel()} require.NoError(t, writeChannelState(kv, 5, defs)) - require.NoError(t, writeHotState(kv, 0, nil, nil, nil)) + require.NoError(t, writeHotState(kv, 0, nil, nil, nil, nil)) cache := protocol.NewChannelCache() s, err := loadKVState(kv, cache) @@ -145,7 +145,7 @@ func Test_KVRecords_DeterministicAndRoundTrip(t *testing.T) { for i := 0; i < 8; i++ { kv := newMemKV() require.NoError(t, writeChannelState(kv, 9, defs)) - require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, carry)) + require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, nil, carry)) if i == 0 { channelBytes, hotBytes = kv.m[string(keyChannelState)], kv.m[string(keyHotState)] continue @@ -156,7 +156,7 @@ func Test_KVRecords_DeterministicAndRoundTrip(t *testing.T) { kv := newMemKV() require.NoError(t, writeChannelState(kv, 9, defs)) - require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, carry)) + require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, nil, carry)) require.Equal(t, uint64(9), binary.BigEndian.Uint64(kv.m[string(keyChannelSeqNr)])) s, err := loadKVState(kv, nil) diff --git a/llo/dev/v31/plugin.go b/llo/dev/v31/plugin.go index a521428..a9266a6 100644 --- a/llo/dev/v31/plugin.go +++ b/llo/dev/v31/plugin.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "time" "golang.org/x/exp/maps" @@ -58,9 +59,17 @@ type Plugin struct { // critical path. Observation only picks up the handle it parked. pump *blobPump + // loggedHistorySkip remembers which channels have already had their + // sampling-rate warning logged, so it is said once per channel per instance + // rather than every cycle. Node-local and for logging only: it must never + // influence anything StateTransition computes, or nodes would diverge. + loggedHistorySkipMu sync.Mutex + loggedHistorySkip map[llotypes.ChannelID]struct{} + // From offchain config - ProtocolVersion uint32 - DefaultMinReportIntervalNanoseconds uint64 + ProtocolVersion uint32 + DefaultMinReportIntervalNanoseconds uint64 + DefaultMinObservationIntervalNanoseconds uint64 } // Query is empty: LLO oracles do not coordinate on what to observe. @@ -86,6 +95,11 @@ func (p *Plugin) Observation(_ context.Context, seqNr uint64, _ ocrtypes.Attribu return nil, fmt.Errorf("failed to load KV state: %w", err) } + obsTSNanos := time.Now().UnixNano() + if obsTSNanos < 0 { + return nil, fmt.Errorf("negative observation timestamps are not supported, got: %d", obsTSNanos) + } + var obs Observation var streams []llotypes.StreamID @@ -110,7 +124,23 @@ func (p *Plugin) Observation(_ context.Context, seqNr uint64, _ ocrtypes.Attribu p.voteOnChannels(&obs, state, seqNr) - streams = observableStreams(state) + if p.DefaultMinObservationIntervalNanoseconds > 0 { + if err := readHotStateForObservation(kvReader, state); err != nil { + return nil, fmt.Errorf("failed to load hot state for observation skip: %w", err) + } + // The hot state lags one round: a channel that reported in the round + // which wrote it has not had its schedule advanced yet, because that + // happens in the next StateTransition. Apply the same advancement + // here so Observation and StateTransition agree on what is due. + for cid, reported := range state.reportedLastRound { + if reported { + state.observationDueNanoseconds[cid] = nextObservationDue( + state.observationDueNanoseconds, cid, + p.DefaultMinObservationIntervalNanoseconds, state.observationTimestampNs) + } + } + } + streams = observableStreams(state, p.DefaultMinObservationIntervalNanoseconds, uint64(obsTSNanos)) } // Stream values are gathered asynchronously by the blob pump and are always @@ -121,40 +151,172 @@ func (p *Plugin) Observation(_ context.Context, seqNr uint64, _ ocrtypes.Attribu // round itself. var handles [][]byte // A nil pump means stream values were never wired up (or the plugin was built - // without the factory); rounds that observe no streams have nothing for the - // pump to gather, so neither publishes input nor consumes a snapshot. - if p.pump != nil && len(streams) > 0 { + // without the factory). + // + // The input is published every round, including rounds that observe nothing, + // so a later cycle can never gather a stale stream set (a cycle with no + // streams parks nothing and is a no-op). Take is called only when this round + // wants values: it consumes and clears the parked snapshot, so calling it on + // a round with nothing to observe would discard a snapshot a later round + // could still use. + if p.pump != nil { p.pump.SetInput(pumpInput{streams: streams, seqNr: seqNr, lifeCycleStage: state.lifeCycleStage}) - if snap, reason := p.pump.Take(seqNr); snap != nil { - handles = append(handles, snap.handleBytes) - } else { - p.Logger.Debugw("No usable stream-value snapshot for this round", "stage", "Observation", "seqNr", seqNr, "reason", reason, "misses", p.pump.Misses(), "cycles", p.pump.Cycles()) + if len(streams) > 0 { + if snap, reason := p.pump.Take(seqNr); snap != nil { + handles = append(handles, snap.handleBytes) + } else { + p.Logger.Debugw("No usable stream-value snapshot for this round", "stage", "Observation", "seqNr", seqNr, "reason", reason, "misses", p.pump.Misses(), "cycles", p.pump.Cycles()) + } } } - obsTSNanos := time.Now().UnixNano() - if obsTSNanos < 0 { - return nil, fmt.Errorf("negative observation timestamps are not supported, got: %d", obsTSNanos) - } obs.UnixTimestampNanoseconds = uint64(obsTSNanos) return encodeObservation(obs, handles) } +// isObservationDue reports whether the channel is due for observation and +// aggregation this round. When minObservationInterval is 0 the feature is +// disabled and every channel is due. +// +// A channel with no schedule entry is due. That covers a channel that has never +// reported, including a newly effective one, which therefore aggregates from its +// first round and builds its initial aggregates and history exactly as it did +// before this interval existed. Entries appear once a channel has reported. +func isObservationDue(observationDue map[llotypes.ChannelID]uint64, channelID llotypes.ChannelID, minObservationInterval, now uint64) bool { + if minObservationInterval == 0 { + return true + } + dueAt, scheduled := observationDue[channelID] + if !scheduled { + return true + } + return now >= dueAt +} + +// nextObservationDue returns the channel's next due timestamp, given that it +// reported at reportedAt. +// +// The schedule is fixed-rate: it advances from the channel's own previous due +// timestamp rather than from the round that reported. That distinction is the +// whole point. A channel's stream values are gathered asynchronously and arrive +// a round after its streams enter the pump's input, so the first due round after +// a skip window withholds and the report lands a round late. Advancing from the +// report would fold that delay into every later cycle and the cadence would +// creep; advancing from the schedule makes it a one-time phase offset instead. +// +// The offset is also what supplies the lead the pump needs: because the schedule +// runs ahead of the watermark by it, a channel becomes due for observation that +// far before it is allowed to report, so its values are gathered by the time it +// reports. The lead is therefore however much the data source actually needs, +// and is not configured anywhere. +// +// This is deliberately not validAfter. That watermark is a report boundary, +// emitted in the report and defining the window (validAfter, observationTimestamp] +// that consecutive reports must tile exactly; anchoring it to a schedule would +// leave gaps between reports. +func nextObservationDue(observationDue map[llotypes.ChannelID]uint64, channelID llotypes.ChannelID, minObservationInterval, reportedAt uint64) uint64 { + prevDue, scheduled := observationDue[channelID] + if !scheduled { + // First report: start the schedule from it. + return reportedAt + minObservationInterval + } + if next := prevDue + minObservationInterval; next > reportedAt { + return next + } + // More than one interval behind, so the channel was unable to report for a + // while. Skip the missed slots rather than firing every round to catch up, + // while staying on the original phase. + missed := (reportedAt - prevDue) / minObservationInterval + return prevDue + (missed+1)*minObservationInterval +} + +// exemptFromObservationSkip reports whether a channel must be observed and +// aggregated every round no matter what its schedule says. +// +// Only history_backfill is: its watermark is a history timestamp rather than a +// report time, so a report cadence means nothing for it. +// +// Channels reading History(...) are deliberately NOT exempt. The skip makes a +// channel's report cadence the sampling rate for its windows, which lowers their +// resolution but does not make them wrong - records carry their own observation +// timestamp, and TWAP integrates over real time. Nor can it stall silently: an +// unreadable window leaves the channel unreportable, which also stops its +// schedule advancing, so it reverts to observing every round until the window is +// satisfied. See DefaultMinObservationIntervalNanoseconds for how to size a +// window against the interval. +func exemptFromObservationSkip(cd llotypes.ChannelDefinition) bool { + return cd.ReportFormat == llotypes.ReportFormatHistoryBackfill +} + +// warnHistorySampledAtReportCadence says once per channel that a channel reading +// stream history is being skipped, so its windows are now sampled at its report +// cadence rather than at the round rate. Whether that is fine depends on the +// depths and thresholds the channel was configured with, which this cannot know, +// so it reports the fact and the interval and leaves the arithmetic to whoever +// reads it. See DefaultMinObservationIntervalNanoseconds. +func (p *Plugin) warnHistorySampledAtReportCadence(channelID llotypes.ChannelID, seqNr uint64) { + p.loggedHistorySkipMu.Lock() + if p.loggedHistorySkip == nil { + p.loggedHistorySkip = map[llotypes.ChannelID]struct{}{} + } + _, said := p.loggedHistorySkip[channelID] + if !said { + p.loggedHistorySkip[channelID] = struct{}{} + } + p.loggedHistorySkipMu.Unlock() + if said { + return + } + p.Logger.Infow("Channel reads stream history and is now sampled at its report cadence, not the round rate; check its history depths and any TWAP thresholds against the observation interval", + "channelID", channelID, + "minObservationIntervalNanoseconds", p.DefaultMinObservationIntervalNanoseconds, + "stage", "StateTransition", "seqNr", seqNr) +} + +// observableDefinitions returns the subset of defs whose channels are due for +// observation/aggregation. Tombstoned and history_backfill channels are always +// retained, as are the channels exemptFromObservationSkip names. That keeps the +// set handed to aggregate and ProcessCalculatedStreams the same whether or not +// the interval is configured. When minObservationInterval is 0, defs is returned +// unchanged. +func observableDefinitions(defs llotypes.ChannelDefinitions, observationDue map[llotypes.ChannelID]uint64, minObservationInterval, now uint64) llotypes.ChannelDefinitions { + if minObservationInterval == 0 { + return defs + } + filtered := make(llotypes.ChannelDefinitions, len(defs)) + for channelID, cd := range defs { + if cd.Tombstone || exemptFromObservationSkip(cd) || + isObservationDue(observationDue, channelID, minObservationInterval, now) { + filtered[channelID] = cd + } + } + return filtered +} + // observableStreams lists the streams a round should observe: every stream of // every live channel, minus calculated streams (which are derived in // StateTransition rather than observed). -func observableStreams(state *kvState) []llotypes.StreamID { +// +// When minObservationInterval is non-zero, channels that are not yet due on the +// observation schedule are skipped: their streams are not observed unless shared +// with a channel that is due or exempt (see exemptFromObservationSkip). A +// channel with no schedule entry yet is always considered due. +func observableStreams(state *kvState, minObservationInterval uint64, now uint64) []llotypes.StreamID { if len(state.channelDefinitions) == 0 { return nil } seen := make(map[llotypes.StreamID]struct{}) streams := make([]llotypes.StreamID, 0, len(state.channelDefinitions)) - for _, cd := range state.channelDefinitions { + for channelID, cd := range state.channelDefinitions { if cd.Tombstone { continue } + if !exemptFromObservationSkip(cd) && + !isObservationDue(state.observationDueNanoseconds, channelID, minObservationInterval, now) { + continue + } for _, strm := range cd.Streams { if strm.Aggregator == llotypes.AggregatorCalculated { continue diff --git a/llo/dev/v31/plugin_test.go b/llo/dev/v31/plugin_test.go index c2b59ae..2ae3b16 100644 --- a/llo/dev/v31/plugin_test.go +++ b/llo/dev/v31/plugin_test.go @@ -72,9 +72,10 @@ func kvChannelDefs(t *testing.T, kv *memKV) llotypes.ChannelDefinitions { func kvHotState(t *testing.T, kv *memKV) *kvState { t.Helper() s := &kvState{ - validAfterNanoseconds: map[llotypes.ChannelID]uint64{}, - reportedLastRound: map[llotypes.ChannelID]bool{}, - carryForward: map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{}, + validAfterNanoseconds: map[llotypes.ChannelID]uint64{}, + reportedLastRound: map[llotypes.ChannelID]bool{}, + observationDueNanoseconds: map[llotypes.ChannelID]uint64{}, + carryForward: map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{}, } require.NoError(t, readHotState(kv, s)) return s @@ -104,15 +105,16 @@ func newFakeBroadcaster() *llotest.BlobBroadcastFetcher { return llotest.NewBlob func testPlugin(t *testing.T) *Plugin { return &Plugin{ - Config: Config{VerboseLogging: true}, - ConfigDigest: ocrtypes.ConfigDigest{1, 2, 3}, - Logger: logger.Test(t), - N: 4, - F: 1, - ReportCodecs: map[llotypes.ReportFormat]protocol.ReportCodec{llotypes.ReportFormatJSON: reportcodec.JSONReportCodec{}}, - ChannelCache: protocol.NewChannelCache(), - ProtocolVersion: 0, - DefaultMinReportIntervalNanoseconds: 0, + Config: Config{VerboseLogging: true}, + ConfigDigest: ocrtypes.ConfigDigest{1, 2, 3}, + Logger: logger.Test(t), + N: 4, + F: 1, + ReportCodecs: map[llotypes.ReportFormat]protocol.ReportCodec{llotypes.ReportFormatJSON: reportcodec.JSONReportCodec{}}, + ChannelCache: protocol.NewChannelCache(), + ProtocolVersion: 0, + DefaultMinReportIntervalNanoseconds: 0, + DefaultMinObservationIntervalNanoseconds: 0, } } @@ -429,7 +431,7 @@ func Test_SecondsResolutionOverlap(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { p := mkPrec(tc.format, tc.opts, tc.validAfter, tc.obsTs) - got := p.reportableChannels(0, protocol.NewOptsCache(), logger.Test(t)) + got := p.reportableChannels(0, 0, protocol.NewOptsCache(), logger.Test(t)) if tc.reportable { require.Equal(t, []llotypes.ChannelID{1}, got) } else { @@ -457,14 +459,14 @@ func Test_DisableNilStreamValues(t *testing.T) { // Missing stream 200 -> not reportable. missing := base(protocol.StreamAggregates{100: {llotypes.AggregatorMedian: protocol.ToDecimal(decimal.NewFromInt(1))}}) - require.Empty(t, missing.reportableChannels(0, protocol.NewOptsCache(), logger.Test(t))) + require.Empty(t, missing.reportableChannels(0, 0, protocol.NewOptsCache(), logger.Test(t))) // Both streams present -> reportable. full := base(protocol.StreamAggregates{ 100: {llotypes.AggregatorMedian: protocol.ToDecimal(decimal.NewFromInt(1))}, 200: {llotypes.AggregatorMedian: protocol.ToDecimal(decimal.NewFromInt(2))}, }) - require.Equal(t, []llotypes.ChannelID{1}, full.reportableChannels(0, protocol.NewOptsCache(), logger.Test(t))) + require.Equal(t, []llotypes.ChannelID{1}, full.reportableChannels(0, 0, protocol.NewOptsCache(), logger.Test(t))) } func Test_DisableNilStreamValues_CalculatedStreams(t *testing.T) { @@ -521,17 +523,17 @@ func Test_DisableNilStreamValues_CalculatedStreams(t *testing.T) { // ProcessCalculatedStreams bailed before writing the calculated // aggregate; the definition alone looks complete. o := mkPrec(true, validOpts, baseStreams, baseAggregates()) - require.Empty(t, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Empty(t, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("inline calculated stream but nil aggregate -> not reportable", func(t *testing.T) { o := mkPrec(true, validOpts, withCalculated, baseAggregates()) - require.Empty(t, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Empty(t, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("fully evaluated -> reportable", func(t *testing.T) { o := mkPrec(true, validOpts, withCalculated, evaluatedAggregates()) - require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("DisableNilStreamValues=false, evaluation failed -> not reportable", func(t *testing.T) { @@ -541,32 +543,32 @@ func Test_DisableNilStreamValues_CalculatedStreams(t *testing.T) { // report. Treating the channel as reportable would advance validAfter // over a round that emitted nothing. o := mkPrec(false, validOpts, baseStreams, baseAggregates()) - require.Empty(t, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Empty(t, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("DisableNilStreamValues=false, fully evaluated -> reportable", func(t *testing.T) { o := mkPrec(false, validOpts, withCalculated, evaluatedAggregates()) - require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("malformed opts -> not reportable", func(t *testing.T) { o := mkPrec(true, []byte(`{"abi":`), withCalculated, evaluatedAggregates()) - require.Empty(t, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Empty(t, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("opts declare no expressions -> not reportable", func(t *testing.T) { o := mkPrec(true, []byte(`{"abi":[]}`), withCalculated, evaluatedAggregates()) - require.Empty(t, o.reportableChannels(0, populatedCache(o), logger.Test(t))) + require.Empty(t, o.reportableChannels(0, 0, populatedCache(o), logger.Test(t))) }) t.Run("cache miss falls back to channel opts -> reportable", func(t *testing.T) { o := mkPrec(true, validOpts, withCalculated, evaluatedAggregates()) - require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(0, protocol.NewOptsCache(), logger.Test(t))) + require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(0, 0, protocol.NewOptsCache(), logger.Test(t))) }) t.Run("cache miss falls back to channel opts -> not reportable when unevaluated", func(t *testing.T) { o := mkPrec(true, validOpts, baseStreams, baseAggregates()) - require.Empty(t, o.reportableChannels(0, protocol.NewOptsCache(), logger.Test(t))) + require.Empty(t, o.reportableChannels(0, 0, protocol.NewOptsCache(), logger.Test(t))) }) } @@ -583,7 +585,7 @@ func Test_TimestampedAggregate_CarryForward(t *testing.T) { obs := map[llotypes.StreamID][]protocol.StreamValue{100: {tsv(ts, v), tsv(ts, v), tsv(ts, v)}} next := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{} // No history requirements: this test is about carry-forward aggregation. - require.NoError(t, p.aggregate(carry, next, defs, obs, out, nil, historyRequirements{}, ts)) + require.NoError(t, p.aggregate(carry, next, defs, defs, obs, out, nil, historyRequirements{}, ts)) carry = next res, ok := out[100][llotypes.AggregatorMedian].(*protocol.TimestampedStreamValue) require.True(t, ok, "expected a TimestampedStreamValue aggregate") @@ -949,3 +951,309 @@ func Test_Observation_RejectsInlineStreamValues(t *testing.T) { var bfErr *blobFetchError require.NotErrorAs(t, err, &bfErr) } + +func Test_ObservationIntervalSkip_observableStreams(t *testing.T) { + state := &kvState{ + channelDefinitions: llotypes.ChannelDefinitions{ + 1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, + 2: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 200, Aggregator: llotypes.AggregatorMedian}}}, + 3: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 300, Aggregator: llotypes.AggregatorMedian}}}, + }, + observationDueNanoseconds: map[llotypes.ChannelID]uint64{ + 1: 4000, + 2: 8000, + }, + } + const interval = 3000 + + // ch3 has never reported, so it has no schedule entry and is always due. + require.ElementsMatch(t, []llotypes.StreamID{300}, observableStreams(state, interval, 3500)) + require.ElementsMatch(t, []llotypes.StreamID{100, 300}, observableStreams(state, interval, 4000), "due exactly at its slot") + require.ElementsMatch(t, []llotypes.StreamID{100, 200, 300}, observableStreams(state, interval, 8500)) + + // interval=0 disables the skip entirely. + require.ElementsMatch(t, []llotypes.StreamID{100, 200, 300}, observableStreams(state, 0, 3500)) + + // A not-due channel sharing a stream with a due one still gets it observed. + state.channelDefinitions[4] = llotypes.ChannelDefinition{ + ReportFormat: llotypes.ReportFormatJSON, + Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}, + } + state.observationDueNanoseconds[4] = 8000 + require.ElementsMatch(t, []llotypes.StreamID{100, 300}, observableStreams(state, interval, 4000)) + + // A stream exclusive to a not-due channel is not observed. + state.channelDefinitions[5] = llotypes.ChannelDefinition{ + ReportFormat: llotypes.ReportFormatJSON, + Streams: []llotypes.Stream{{StreamID: 500, Aggregator: llotypes.AggregatorMedian}}, + } + state.observationDueNanoseconds[5] = 8000 + require.ElementsMatch(t, []llotypes.StreamID{100, 300}, observableStreams(state, interval, 4000)) +} + +// The persisted schedule lags one round behind: a channel that reported in the +// round which wrote the hot state has not had its schedule advanced yet. +// Observation applies that advancement itself so it agrees with StateTransition. +func Test_ObservationIntervalSkip_ObservationAdvancesLaggingSchedule(t *testing.T) { + const interval = 5000 + state := &kvState{ + channelDefinitions: llotypes.ChannelDefinitions{ + 1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, + }, + observationDueNanoseconds: map[llotypes.ChannelID]uint64{1: 8000}, + reportedLastRound: map[llotypes.ChannelID]bool{1: true}, + observationTimestampNs: 8020, + } + // As persisted, the channel is still due at its old slot. + require.ElementsMatch(t, []llotypes.StreamID{100}, observableStreams(state, interval, 8020)) + + for cid, reported := range state.reportedLastRound { + if reported { + state.observationDueNanoseconds[cid] = nextObservationDue( + state.observationDueNanoseconds, cid, interval, state.observationTimestampNs) + } + } + + // Advanced from the schedule (8000+5000), not from the report (8020+5000): + // the 20ns the report ran late does not move the next slot. + require.Equal(t, uint64(13000), state.observationDueNanoseconds[1]) + require.Empty(t, observableStreams(state, interval, 10000)) + require.ElementsMatch(t, []llotypes.StreamID{100}, observableStreams(state, interval, 13000)) +} + +func Test_nextObservationDue(t *testing.T) { + const interval = 1000 + + // No entry yet: the first report starts the schedule. + require.Equal(t, uint64(2020), nextObservationDue(map[llotypes.ChannelID]uint64{}, 1, interval, 1020)) + + // Fixed-rate: a report that ran 20 late still advances by exactly one + // interval from the schedule, so the lateness does not accumulate. + sched := map[llotypes.ChannelID]uint64{1: 1000} + require.Equal(t, uint64(2000), nextObservationDue(sched, 1, interval, 1020)) + + // Steady state: the phase offset stays put over many cycles rather than + // creeping, which is the whole point of anchoring to the schedule. + due, reportedAt := uint64(1000), uint64(1020) + for range 100 { + sched[1] = due + due = nextObservationDue(sched, 1, interval, reportedAt) + reportedAt = due + 20 // always one round late + } + require.Equal(t, uint64(101000), due, "cadence must not drift") + + // Far behind: skip the missed slots instead of firing every round to catch + // up, staying on the original phase. + sched[1] = 1000 + require.Equal(t, uint64(5000), nextObservationDue(sched, 1, interval, 4020)) + require.Greater(t, nextObservationDue(sched, 1, interval, 4020), uint64(4020)) +} + +// A channel that is due but whose stream values have not arrived yet (the pump +// serves a round from the snapshot gathered under the previous round's stream +// set, so the first due round after a skip window has none) must withhold its +// report rather than emit nils. Withholding leaves validAfter where it is, so +// the channel stays due and reports on the following round instead. +func Test_ObservationIntervalSkip_WithholdsReportUntilValuesArrive(t *testing.T) { + const interval = 5000 + + o := precursor{ + LifeCycleStage: protocol.LifeCycleStageProduction, + ChannelDefinitions: llotypes.ChannelDefinitions{ + 1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, + }, + ValidAfterNanoseconds: map[llotypes.ChannelID]uint64{1: 1000}, + ObservationTimestampNanoseconds: 6000, + StreamAggregates: protocol.StreamAggregates{}, + } + cache := protocol.NewOptsCache() + + // Due on time (6000 >= 1000+5000) but the aggregate is missing. + require.Empty(t, o.reportableChannels(interval, interval, cache, logger.Test(t)), + "a channel with no aggregate must not report") + + // The next round, the pump has delivered and the channel reports. + o.StreamAggregates[100] = map[llotypes.Aggregator]protocol.StreamValue{ + llotypes.AggregatorMedian: protocol.ToDecimal(decimal.NewFromInt(42)), + } + require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(interval, interval, cache, logger.Test(t)), + "once values arrive the channel reports") + + // With the feature disabled the rule does not apply, so behaviour for + // existing deployments is unchanged. + delete(o.StreamAggregates, 100) + require.Equal(t, []llotypes.ChannelID{1}, o.reportableChannels(interval, 0, cache, logger.Test(t)), + "the withholding rule is gated on the observation interval") +} + +// A pair belonging only to a skipped channel keeps its carried value, so the +// channel does not come out of a skip window worse off than it went in: the +// last-known-good value is still there as the fallback on the round it returns. +// A pair whose last live channel has gone is still reclaimed. +func Test_ObservationIntervalSkip_CarryForwardSurvivesSkip(t *testing.T) { + p := testPlugin(t) + const interval = 5000 + + defs := llotypes.ChannelDefinitions{ + 1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, + } + // Not due until 10000. + schedule := map[llotypes.ChannelID]uint64{1: 10_000} + carried := &protocol.TimestampedStreamValue{ObservedAtNanoseconds: 42, StreamValue: protocol.ToDecimal(decimal.NewFromInt(7))} + prevCarry := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{ + 100: {llotypes.AggregatorMedian: carried}, + } + + // Round where channel 1 is skipped: no observations for stream 100 at all. + next := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{} + out := protocol.StreamAggregates{} + due := observableDefinitions(defs, schedule, interval, 6000) + require.Empty(t, due, "channel must be skipped for this test to mean anything") + + require.NoError(t, p.aggregate(prevCarry, next, due, defs, nil, out, nil, historyRequirements{}, 6000)) + + require.Same(t, carried, next[100][llotypes.AggregatorMedian], + "a skipped channel's carried value must survive the round") + require.Nil(t, out[100][llotypes.AggregatorMedian], + "but it must not be published as this round's aggregate, which would make the channel reportable") + + // Same round with the channel no longer live: the value is reclaimed. + orphaned := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{} + require.NoError(t, p.aggregate(prevCarry, orphaned, llotypes.ChannelDefinitions{}, llotypes.ChannelDefinitions{}, + nil, protocol.StreamAggregates{}, nil, historyRequirements{}, 6000)) + require.Empty(t, orphaned, "a pair no live channel declares must not be carried forward") +} + +func Test_ObservationIntervalSkip_aggregate(t *testing.T) { + p := testPlugin(t) + p.DefaultMinObservationIntervalNanoseconds = 5000 + + defs := llotypes.ChannelDefinitions{ + 1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, + 2: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 200, Aggregator: llotypes.AggregatorMedian}}}, + } + // Observation schedule: ch1's slot has already passed, ch2's has not. + schedule := map[llotypes.ChannelID]uint64{1: 1000, 2: 10000} + + mkObs := func(sid llotypes.StreamID, v int64) []protocol.StreamValue { + return []protocol.StreamValue{ + protocol.ToDecimal(decimal.NewFromInt(v)), + protocol.ToDecimal(decimal.NewFromInt(v)), + protocol.ToDecimal(decimal.NewFromInt(v)), + } + } + obs := map[llotypes.StreamID][]protocol.StreamValue{ + 100: mkObs(100, 42), + 200: mkObs(200, 99), + } + + out := protocol.StreamAggregates{} + next := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{} + + // t=6000: ch1 due (6000 >= its slot at 1000), ch2 not (6000 < its slot at 10000). + err := p.aggregate(nil, next, observableDefinitions(defs, schedule, 5000, 6000), defs, obs, out, nil, historyRequirements{}, 6000) + require.NoError(t, err) + + require.NotNil(t, out[100][llotypes.AggregatorMedian], "due channel's stream must be aggregated") + require.Nil(t, out[200][llotypes.AggregatorMedian], "not-due channel's stream must be skipped") + + // interval=0: all channels aggregated (disabled) + out2 := protocol.StreamAggregates{} + next2 := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{} + err = p.aggregate(nil, next2, observableDefinitions(defs, schedule, 0, 6000), defs, obs, out2, nil, historyRequirements{}, 6000) + require.NoError(t, err) + require.NotNil(t, out2[100][llotypes.AggregatorMedian]) + require.NotNil(t, out2[200][llotypes.AggregatorMedian]) +} + +func Test_ObservationIntervalSkip_FullRound(t *testing.T) { + ctx := tests.Context(t) + p := testPlugin(t) + p.DefaultMinReportIntervalNanoseconds = 5000 + p.DefaultMinObservationIntervalNanoseconds = 5000 + kv := newMemKV() + + channelDef := llotypes.ChannelDefinition{ + ReportFormat: llotypes.ReportFormatJSON, + Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}, + } + + obsWithVal := func(ts uint64, v int64) []ocrtypes.AttributedObservation { + obs := Observation{ + UnixTimestampNanoseconds: ts, + StreamValues: protocol.StreamValues{100: protocol.ToDecimal(decimal.NewFromInt(v))}, + } + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + aos = append(aos, ao(i, mustEncodeObs(t, obs))) + } + return aos + } + + // Round 1: bootstrap + _, err := p.StateTransition(ctx, 1, ocrtypes.AttributedQuery{}, []ocrtypes.AttributedObservation{ao(0, nil), ao(1, nil), ao(2, nil)}, kv, testBlobs) + require.NoError(t, err) + + // Round 2: add channel 1 + _, err = p.StateTransition(ctx, 2, ocrtypes.AttributedQuery{}, addChannelRound(t, 1_000, 1, channelDef), kv, testBlobs) + require.NoError(t, err) + require.Contains(t, kvChannelDefs(t, kv), llotypes.ChannelID(1)) + + // Round 3: channel effective, first watermark. It has never reported, so it + // has no schedule slot and is due: it aggregates from its very first round + // and builds its initial aggregates, exactly as it did before this interval + // existed. It is still not reportable, because validAfter == now. + prec3, err := p.StateTransition(ctx, 3, ocrtypes.AttributedQuery{}, obsWithVal(3_000, 10), kv, testBlobs) + require.NoError(t, err) + p3, err := decodePrecursor(prec3) + require.NoError(t, err) + require.NotNil(t, p3.StreamAggregates[100][llotypes.AggregatorMedian], + "round 3: a channel that has never reported is due and must be aggregated") + reports3, err := p.Reports(ctx, 3, prec3) + require.NoError(t, err) + require.Empty(t, reports3, "round 3: no reports") + assert.Equal(t, uint64(3000), storedValidAfter(t, kv, 1)) + require.False(t, reportedFlag(t, kv, 1)) + require.Zero(t, storedObservationDue(t, kv, 1), "round 3: not scheduled until it has reported") + + // Round 4: still unscheduled, so still due. Aggregated and reported. + prec4, err := p.StateTransition(ctx, 4, ocrtypes.AttributedQuery{}, obsWithVal(8_000, 20), kv, testBlobs) + require.NoError(t, err) + p4, err := decodePrecursor(prec4) + require.NoError(t, err) + require.NotNil(t, p4.StreamAggregates[100][llotypes.AggregatorMedian], + "round 4: channel due, stream must be aggregated") + reports4, err := p.Reports(ctx, 4, prec4) + require.NoError(t, err) + require.Len(t, reports4, 1, "round 4: one report") + assert.Equal(t, uint64(3000), storedValidAfter(t, kv, 1), "validAfter not yet advanced (advances next round)") + require.True(t, reportedFlag(t, kv, 1)) + + // Round 5: the previous round reported, so the schedule is seeded to 13000 + // and validAfter advances to 8000. Not due: 10000 < 13000. + prec5, err := p.StateTransition(ctx, 5, ocrtypes.AttributedQuery{}, obsWithVal(10_000, 30), kv, testBlobs) + require.NoError(t, err) + p5, err := decodePrecursor(prec5) + require.NoError(t, err) + require.Nil(t, p5.StreamAggregates[100][llotypes.AggregatorMedian], + "round 5: channel not due, stream must not be aggregated") + reports5, err := p.Reports(ctx, 5, prec5) + require.NoError(t, err) + require.Empty(t, reports5, "round 5: no reports") + assert.Equal(t, uint64(8000), storedValidAfter(t, kv, 1), "validAfter advanced because prev round reported") + require.False(t, reportedFlag(t, kv, 1)) + assert.Equal(t, uint64(13_000), storedObservationDue(t, kv, 1), + "schedule seeded from the round that reported (8000+5000)") + + // Round 6: due again, 14000 >= its slot at 13000. Aggregated and reported. + prec6, err := p.StateTransition(ctx, 6, ocrtypes.AttributedQuery{}, obsWithVal(14_000, 40), kv, testBlobs) + require.NoError(t, err) + p6, err := decodePrecursor(prec6) + require.NoError(t, err) + require.NotNil(t, p6.StreamAggregates[100][llotypes.AggregatorMedian], + "round 6: channel due, stream must be aggregated") + reports6, err := p.Reports(ctx, 6, prec6) + require.NoError(t, err) + require.Len(t, reports6, 1, "round 6: one report") + require.True(t, reportedFlag(t, kv, 1)) +} diff --git a/llo/dev/v31/reports.go b/llo/dev/v31/reports.go index dc82d1c..04fbe4b 100644 --- a/llo/dev/v31/reports.go +++ b/llo/dev/v31/reports.go @@ -62,7 +62,7 @@ func (p *Plugin) Reports(ctx context.Context, seqNr uint64, rawPrecursor ocr3_1t }) } - for _, cid := range out.reportableChannels(p.DefaultMinReportIntervalNanoseconds, channelOpts, p.Logger) { + for _, cid := range out.reportableChannels(p.DefaultMinReportIntervalNanoseconds, p.DefaultMinObservationIntervalNanoseconds, channelOpts, p.Logger) { cd := out.ChannelDefinitions[cid] if cd.ReportFormat == llotypes.ReportFormatHistoryBackfill { @@ -179,10 +179,10 @@ func (p *Plugin) Reports(ctx context.Context, seqNr uint64, rawPrecursor ocr3_1t // reportableChannels returns the sorted set of channels reportable in this // (current) round (see isReportable). -func (o precursor) reportableChannels(minReportInterval uint64, optsCache *protocol.OptsCache, lggr logger.Logger) []llotypes.ChannelID { +func (o precursor) reportableChannels(minReportInterval, minObservationInterval uint64, optsCache *protocol.OptsCache, lggr logger.Logger) []llotypes.ChannelID { reportable := make([]llotypes.ChannelID, 0, len(o.ChannelDefinitions)) for channelID := range o.ChannelDefinitions { - if o.isReportable(channelID, minReportInterval, optsCache, lggr) { + if o.isReportable(channelID, minReportInterval, minObservationInterval, optsCache, lggr) { reportable = append(reportable, channelID) } } @@ -190,7 +190,7 @@ func (o precursor) reportableChannels(minReportInterval uint64, optsCache *proto return reportable } -func (o precursor) isReportable(channelID llotypes.ChannelID, minReportInterval uint64, optsCache *protocol.OptsCache, lggr logger.Logger) bool { +func (o precursor) isReportable(channelID llotypes.ChannelID, minReportInterval, minObservationInterval uint64, optsCache *protocol.OptsCache, lggr logger.Logger) bool { if o.LifeCycleStage == protocol.LifeCycleStageRetired { return false } @@ -202,6 +202,31 @@ func (o precursor) isReportable(channelID llotypes.ChannelID, minReportInterval _, _, _, ok := selectBackfillCandidate(o.ChannelDefinitions, o.ValidAfterNanoseconds, o.ObservationTimestampNanoseconds, channelID, optsCache) return ok } + // The observation interval skips gathering a not-due channel's streams, and + // the blob pump serves a round from the snapshot gathered under the previous + // round's stream set. So on the round a channel first becomes due its values + // have not arrived yet and it has no aggregates. Reporting anyway would emit + // nil values and advance validAfter over a round that carried nothing. + // + // Withholding instead leaves both the watermark and the observation schedule + // where they are, because each only advances on a round that reported: the + // channel stays due, its streams are in the pump's input now, and it reports + // once they arrive. The schedule then advances from its own slot rather than + // from the late report (see nextObservationDue), so the delay settles into a + // constant phase offset instead of being added to every later cycle - and + // that offset is exactly the lead the pump needs on every later cycle. + if minObservationInterval > 0 && !cd.DisableNilStreamValues { + for _, strm := range cd.Streams { + if strm.Aggregator == llotypes.AggregatorCalculated { + continue + } + if o.StreamAggregates[strm.StreamID][strm.Aggregator] == nil { + lggr.Debugw("IsReportable=false; awaiting stream values after observation skip", + "channelID", channelID, "streamID", strm.StreamID) + return false + } + } + } // When DisableNilStreamValues is set, every stream must have a (non-nil) // aggregate value for the channel to be reportable. if cd.DisableNilStreamValues { diff --git a/llo/dev/v31/statetransition.go b/llo/dev/v31/statetransition.go index 0887a86..bd4f40f 100644 --- a/llo/dev/v31/statetransition.go +++ b/llo/dev/v31/statetransition.go @@ -59,7 +59,7 @@ func (p *Plugin) StateTransition(ctx context.Context, seqNr uint64, _ ocrtypes.A if err := writeChannelState(kvRW, seqNr, nil); err != nil { return nil, err } - if err := writeHotState(kvRW, 0, nil, nil, nil); err != nil { + if err := writeHotState(kvRW, 0, nil, nil, nil, nil); err != nil { return nil, err } return encodePrecursor(precursor{LifeCycleStage: stage}) @@ -184,12 +184,63 @@ func (p *Plugin) StateTransition(ctx context.Context, seqNr uint64, _ ocrtypes.A "streamID", key.streamID, "aggregator", key.aggregator, "seqNr", seqNr) } + // Observation schedule. A channel's next due time advances from its own + // previous due time and only when it actually reported, so a cycle that ran + // late (the round a channel withholds while its stream values are still + // being gathered, or one where aggregation failed) does not push every later + // cycle out with it. See nextObservationDue. + // + // The hot state lags one round, exactly as validAfter does: this round + // advances the schedule of whatever the previous round reported. + // + // Promotion clears the schedule for the same reason it replaces validAfter + // wholesale: a slot inherited from staging could leave a channel not due on + // the promotion round, and a channel that is not aggregated has no values to + // report, which is exactly the gap the handover exists to avoid. An unset + // schedule means due, so every channel aggregates immediately and rebuilds + // its slot from its first report. + observationDue := map[llotypes.ChannelID]uint64{} + if p.DefaultMinObservationIntervalNanoseconds > 0 && promotedValidAfter == nil { + for channelID, cd := range effective { + if cd.Tombstone || exemptFromObservationSkip(cd) { + // Never skipped, so never scheduled. + continue + } + if prevReportable(prev, channelID) { + observationDue[channelID] = nextObservationDue(prev.observationDueNanoseconds, channelID, + p.DefaultMinObservationIntervalNanoseconds, prev.observationTimestampNs) + } else if dueAt, scheduled := prev.observationDueNanoseconds[channelID]; scheduled { + // Did not report: leave the schedule where it is so the channel + // stays due and retries. + observationDue[channelID] = dueAt + } + // Otherwise unscheduled, i.e. due: a channel that has never reported + // aggregates every round, as it did before this interval existed. + } + } + + // When DefaultMinObservationIntervalNanoseconds is configured, only channels + // due on that schedule are aggregated; the rest are skipped to save work on + // channels that will not report this round. The filter is applied once, here, + // and the result drives both aggregation and calculated stream evaluation, so + // the two can never disagree about which channels this round covers. + aggregationDefs := observableDefinitions(effective, observationDue, p.DefaultMinObservationIntervalNanoseconds, out.ObservationTimestampNanoseconds) + + // Node-local diagnostic only; nothing below reads it. + if p.DefaultMinObservationIntervalNanoseconds > 0 { + for channelID, cd := range effective { + if _, due := aggregationDefs[channelID]; !due && protocol.HasCalculatedStreams(cd) { + p.warnHistorySampledAtReportCadence(channelID, seqNr) + } + } + } + // Aggregation (regular fresh; timestamped with cross-round carry-forward via // the r/agg record). carryForward accumulates the values to persist for the - // next round. Runs over the effective set, which is what was observed. The + // next round. Runs over the filtered set, which is what was observed. The // agreed value of every pair history requires is recorded as it is computed. carryForward := map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{} - if err := p.aggregate(prev.carryForward, carryForward, effective, streamObservations, out.StreamAggregates, + if err := p.aggregate(prev.carryForward, carryForward, aggregationDefs, effective, streamObservations, out.StreamAggregates, history, requirements, out.ObservationTimestampNanoseconds); err != nil { return nil, err } @@ -204,10 +255,13 @@ func (p *Plugin) StateTransition(ctx context.Context, seqNr uint64, _ ocrtypes.A // channel reports is derived from its opts by protocol.EffectiveStreams, so // nothing about evaluation reaches replicated state and a persisted // definition stays exactly what was voted on. - calculated.ProcessCalculatedStreams(p.Logger, effective, out.StreamAggregates, out.ObservationTimestampNanoseconds, prev.opts, history) + // + // Runs over the same filtered set as aggregation: calculated streams for + // not-due channels are not needed (the channel will not report). + calculated.ProcessCalculatedStreams(p.Logger, aggregationDefs, out.StreamAggregates, out.ObservationTimestampNanoseconds, prev.opts, history) // Flush KV mutations. - if err := p.flushKV(kvRW, seqNr, prev, out, pending, carryForward, history); err != nil { + if err := p.flushKV(kvRW, seqNr, prev, out, pending, observationDue, carryForward, history); err != nil { return nil, err } @@ -366,6 +420,7 @@ func applyChannelVotes( func (p *Plugin) aggregate( prevCarry, nextCarry map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue, defs llotypes.ChannelDefinitions, + liveDefs llotypes.ChannelDefinitions, streamObservations map[llotypes.StreamID][]protocol.StreamValue, out protocol.StreamAggregates, history *historyStore, @@ -379,6 +434,14 @@ func (p *Plugin) aggregate( nextCarry[sid][agg] = tsv } + // defs has already been narrowed to the channels due this round (see + // observableDefinitions). Pairs belonging only to channels that were filtered + // out are carried forward untouched by the loop below this one. + // Pairs this round actually looked at. A pair that was looked at owns its own + // carry-forward outcome, including the deliberate drops below, so the + // preservation pass at the end must not second-guess it. + visited := map[histKey]struct{}{} + for _, cd := range defs { if cd.Tombstone || cd.ReportFormat == llotypes.ReportFormatHistoryBackfill { // Not aggregated, so nothing is carried forward on their behalf. A @@ -394,6 +457,7 @@ func (p *Plugin) aggregate( // are recomputed each round by ProcessCalculatedStreams. continue } + visited[histKey{streamID: sid, aggregator: agg}] = struct{}{} if _, exists := out[sid][agg]; exists { continue } @@ -452,6 +516,42 @@ func (p *Plugin) aggregate( } } } + + // Pairs belonging only to channels the observation schedule skipped were + // never looked at above. They keep their carried value, exactly as a pair + // keeps it on a round where aggregation failed: skipping a channel must not + // cost it the last-known-good value behind timestamped monotonicity, or it + // could adopt an older value coming out of a skip window than it held going + // in, and would have no fallback on the round it returns. + // + // Driven from liveDefs, not from prevCarry, so a pair whose last channel has + // been removed is still reclaimed by not being written into the new hot + // record. Restricted to unvisited pairs, so the deliberate drops above - a + // pair that turned non-timestamped, a transient failure with nothing carried + // - keep their decision. + // + // This does not reach history: appendHistory is not called here, and its + // strictly-newer guard would reject the value anyway. That is deliberate - a + // carried value must not be counted once per round in a window, which is why + // history records a gap rather than a repeat. Nor does it make a skipped + // channel reportable: the value goes into nextCarry only, never into out. + for _, cd := range liveDefs { + if cd.Tombstone || cd.ReportFormat == llotypes.ReportFormatHistoryBackfill { + continue + } + for _, strm := range cd.Streams { + sid, agg := strm.StreamID, strm.Aggregator + if agg == llotypes.AggregatorCalculated { + continue + } + if _, seen := visited[histKey{streamID: sid, aggregator: agg}]; seen { + continue + } + if tsv := prevCarry[sid][agg]; tsv != nil { + keep(sid, agg, tsv) + } + } + } return nil } @@ -497,6 +597,7 @@ func (p *Plugin) flushKV( prev *kvState, out precursor, pending llotypes.ChannelDefinitions, + observationDue map[llotypes.ChannelID]uint64, carryForward map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue, history *historyStore, ) error { @@ -522,7 +623,7 @@ func (p *Plugin) flushKV( // round can advance validAfter faithfully (see prevReportable). reportable := make(map[llotypes.ChannelID]bool, len(out.ChannelDefinitions)) for id := range out.ChannelDefinitions { - reportable[id] = out.isReportable(id, p.DefaultMinReportIntervalNanoseconds, prev.opts, p.Logger) + reportable[id] = out.isReportable(id, p.DefaultMinReportIntervalNanoseconds, p.DefaultMinObservationIntervalNanoseconds, prev.opts, p.Logger) } // Stream history: write modified windows, delete pairs no live channel @@ -533,7 +634,7 @@ func (p *Plugin) flushKV( } } - return writeHotState(kvRW, out.ObservationTimestampNanoseconds, out.ValidAfterNanoseconds, reportable, carryForward) + return writeHotState(kvRW, out.ObservationTimestampNanoseconds, out.ValidAfterNanoseconds, reportable, observationDue, carryForward) } // channelDefinitionsChanged reports whether the channel set or any individual diff --git a/llo/protocol/calculated/doc.go b/llo/protocol/calculated/doc.go index 13baf24..209b1d5 100644 --- a/llo/protocol/calculated/doc.go +++ b/llo/protocol/calculated/doc.go @@ -84,6 +84,20 @@ // the change as a NEW channel, wait for its history to be satisfied, then retire // the old one. Lowering a depth takes effect the next round with no gap. // +// # Sampling rate +// +// A window is appended to on the rounds its channel is aggregated, so the depth +// needed to cover a given span of wall clock depends on how often that happens. +// Where DefaultMinObservationIntervalNanoseconds is configured, that is the +// channel's report cadence rather than the round rate, and depths and thresholds +// must be sized against it. +// +// TWAP is the exception that cares least: it buckets its window by the second +// and keeps the newest record per bucket, so sampling faster than 1Hz buys it +// nothing and an interval at or below a second leaves it unchanged. Every other +// window function reads the record series directly, so its wall-clock meaning +// follows the sampling rate at any interval. +// // # Limits // // Depth per (stream, aggregator) pair, the number of such pairs, the per-round diff --git a/llo/protocol/llo_offchain_config.pb.go b/llo/protocol/llo_offchain_config.pb.go index 64a2b07..f94a8eb 100644 --- a/llo/protocol/llo_offchain_config.pb.go +++ b/llo/protocol/llo_offchain_config.pb.go @@ -22,12 +22,13 @@ const ( ) type LLOOffchainConfigProto struct { - state protoimpl.MessageState `protogen:"open.v1"` - ProtocolVersion uint32 `protobuf:"varint,1,opt,name=protocolVersion,proto3" json:"protocolVersion,omitempty"` - DefaultMinReportIntervalNanoseconds uint64 `protobuf:"varint,2,opt,name=defaultMinReportIntervalNanoseconds,proto3" json:"defaultMinReportIntervalNanoseconds,omitempty"` - EnableObservationCompression bool `protobuf:"varint,3,opt,name=enableObservationCompression,proto3" json:"enableObservationCompression,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + ProtocolVersion uint32 `protobuf:"varint,1,opt,name=protocolVersion,proto3" json:"protocolVersion,omitempty"` + DefaultMinReportIntervalNanoseconds uint64 `protobuf:"varint,2,opt,name=defaultMinReportIntervalNanoseconds,proto3" json:"defaultMinReportIntervalNanoseconds,omitempty"` + EnableObservationCompression bool `protobuf:"varint,3,opt,name=enableObservationCompression,proto3" json:"enableObservationCompression,omitempty"` + DefaultMinObservationIntervalNanoseconds uint64 `protobuf:"varint,4,opt,name=defaultMinObservationIntervalNanoseconds,proto3" json:"defaultMinObservationIntervalNanoseconds,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *LLOOffchainConfigProto) Reset() { @@ -81,15 +82,23 @@ func (x *LLOOffchainConfigProto) GetEnableObservationCompression() bool { return false } +func (x *LLOOffchainConfigProto) GetDefaultMinObservationIntervalNanoseconds() uint64 { + if x != nil { + return x.DefaultMinObservationIntervalNanoseconds + } + return 0 +} + var File_llo_offchain_config_proto protoreflect.FileDescriptor const file_llo_offchain_config_proto_rawDesc = "" + "\n" + - "\x19llo_offchain_config.proto\x12\x02v1\"\xd8\x01\n" + + "\x19llo_offchain_config.proto\x12\x02v1\"\xb4\x02\n" + "\x16LLOOffchainConfigProto\x12(\n" + "\x0fprotocolVersion\x18\x01 \x01(\rR\x0fprotocolVersion\x12P\n" + "#defaultMinReportIntervalNanoseconds\x18\x02 \x01(\x04R#defaultMinReportIntervalNanoseconds\x12B\n" + - "\x1cenableObservationCompression\x18\x03 \x01(\bR\x1cenableObservationCompressionB\fZ\n" + + "\x1cenableObservationCompression\x18\x03 \x01(\bR\x1cenableObservationCompression\x12Z\n" + + "(defaultMinObservationIntervalNanoseconds\x18\x04 \x01(\x04R(defaultMinObservationIntervalNanosecondsB\fZ\n" + ".;protocolb\x06proto3" var ( diff --git a/llo/protocol/llo_offchain_config.proto b/llo/protocol/llo_offchain_config.proto index 73b86a2..2b323ec 100644 --- a/llo/protocol/llo_offchain_config.proto +++ b/llo/protocol/llo_offchain_config.proto @@ -7,4 +7,5 @@ message LLOOffchainConfigProto { uint32 protocolVersion = 1; uint64 defaultMinReportIntervalNanoseconds = 2; bool enableObservationCompression = 3; + uint64 defaultMinObservationIntervalNanoseconds = 4; } diff --git a/llo/protocol/offchain_config.go b/llo/protocol/offchain_config.go index b8ee4db..d83afa9 100644 --- a/llo/protocol/offchain_config.go +++ b/llo/protocol/offchain_config.go @@ -18,6 +18,54 @@ type OffchainConfig struct { // produced quickly, you are still limited by OCR3's DeltaRound and // DeltaGrace params, as well as networking latency. DefaultMinReportIntervalNanoseconds uint64 + // DefaultMinObservationIntervalNanoseconds is the default minimum interval + // in nanoseconds between the last report of a channel and the next time its + // streams are observed/aggregated. Each channel carries an observation + // schedule advanced by this interval whenever it reports; until its next + // slot comes round, its observation and aggregation are skipped entirely. + // + // It must be set to 0 for protocol version 0. + // For protocol version 1+, 0 means disabled (all channels are always + // observed); a non-zero value enables the skip. It must not exceed + // DefaultMinReportIntervalNanoseconds, or a channel could be reportable + // but lack the observations needed to produce a report. + // + // Setting it equal to DefaultMinReportIntervalNanoseconds is safe. The first + // round after a skip window has no stream values yet (they are gathered + // asynchronously, so they arrive a round later), and a channel with no + // aggregate withholds its report until they land rather than emitting one + // full of nils. + // + // That delay does not accumulate. The observation schedule advances at a + // fixed rate from its own previous slot rather than from the round that + // reported, so a channel configured to report every T keeps reporting every + // T, offset once by however long its first cycle took to gather. + // + // SIZING HISTORY WINDOWS. Enabling this makes a channel's report cadence the + // sampling rate for any History(...) window it reads, so a window's depth + // buys a different amount of wall-clock coverage than it did at the round + // rate. Size depth, and the TWAP window/minSamples/gap thresholds, against + // this interval. + // + // For TWAP specifically, an interval at or below one second changes nothing: + // TWAP buckets its window by the second and takes the newest record in each + // bucket, so a faster sampling rate was already being discarded. It makes + // depth go further, since depth stops being spent on records that collapse + // into the same bucket. Above one second, the observed bucket count falls to + // about window/interval and every interior gap becomes interval-1 buckets, + // which is where minSamples and maxInteriorGap start to bite. + // + // The other window functions (Avg, Median, EMA, SMA, WMA, Delta, PctChange, + // Spread, Variance, Stddev, Last) read the record series directly with no + // bucketing, so their wall-clock meaning tracks the sampling rate at any + // interval: EMA(History(s, 50), 20) smooths over a very different span at a + // 20ms round than at a 1s cadence. + // + // Getting this wrong costs reports, not correctness. A window that cannot be + // satisfied leaves its channel unreportable, which also stops its schedule + // advancing, so the channel reverts to observing every round until the window + // is satisfied again. + DefaultMinObservationIntervalNanoseconds uint64 // EnableObservationCompression enables observation compression. EnableObservationCompression bool } @@ -42,15 +90,17 @@ func DecodeOffchainConfig(b []byte) (o OffchainConfig, err error) { } o.ProtocolVersion = pbuf.ProtocolVersion o.DefaultMinReportIntervalNanoseconds = pbuf.DefaultMinReportIntervalNanoseconds + o.DefaultMinObservationIntervalNanoseconds = pbuf.DefaultMinObservationIntervalNanoseconds o.EnableObservationCompression = pbuf.EnableObservationCompression return } func (c OffchainConfig) Encode() ([]byte, error) { pbuf := &LLOOffchainConfigProto{ - ProtocolVersion: c.ProtocolVersion, - DefaultMinReportIntervalNanoseconds: c.DefaultMinReportIntervalNanoseconds, - EnableObservationCompression: c.EnableObservationCompression, + ProtocolVersion: c.ProtocolVersion, + DefaultMinReportIntervalNanoseconds: c.DefaultMinReportIntervalNanoseconds, + DefaultMinObservationIntervalNanoseconds: c.DefaultMinObservationIntervalNanoseconds, + EnableObservationCompression: c.EnableObservationCompression, } return proto.Marshal(pbuf) } @@ -61,10 +111,16 @@ func (c OffchainConfig) Validate() error { if c.DefaultMinReportIntervalNanoseconds != 0 { return errors.New("default report cadence must be 0 if protocol version is 0") } + if c.DefaultMinObservationIntervalNanoseconds != 0 { + return errors.New("default observation cadence must be 0 if protocol version is 0") + } case 1: if c.DefaultMinReportIntervalNanoseconds == 0 { return errors.New("default report cadence must be non-zero if protocol version is 1") } + if c.DefaultMinObservationIntervalNanoseconds > c.DefaultMinReportIntervalNanoseconds { + return errors.New("default observation cadence must not exceed default report cadence") + } default: return fmt.Errorf("unknown protocol version: %d", c.ProtocolVersion) } diff --git a/llo/protocol/offchain_config_test.go b/llo/protocol/offchain_config_test.go index 35400a3..d9cdaa4 100644 --- a/llo/protocol/offchain_config_test.go +++ b/llo/protocol/offchain_config_test.go @@ -47,13 +47,24 @@ func Test_OffchainConfig(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "default report cadence must be 0 if protocol version is 0") }) + t.Run("setting DefaultMinObservationIntervalNanoseconds is invalid", func(t *testing.T) { + cfg := OffchainConfig{ + ProtocolVersion: 0, + DefaultMinObservationIntervalNanoseconds: 1, + } + + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "default observation cadence must be 0 if protocol version is 0") + }) }) t.Run("version 1", func(t *testing.T) { t.Run("encode/decode valid values", func(t *testing.T) { cfg := OffchainConfig{ - ProtocolVersion: 1, - DefaultMinReportIntervalNanoseconds: 1000, - EnableObservationCompression: true, + ProtocolVersion: 1, + DefaultMinReportIntervalNanoseconds: 1000, + DefaultMinObservationIntervalNanoseconds: 500, + EnableObservationCompression: true, } b, err := cfg.Encode() @@ -63,6 +74,24 @@ func Test_OffchainConfig(t *testing.T) { require.NoError(t, err) assert.Equal(t, cfg, cfgDecoded) }) + t.Run("DefaultMinObservationIntervalNanoseconds=0 is valid (disabled)", func(t *testing.T) { + cfg := OffchainConfig{ + ProtocolVersion: 1, + DefaultMinReportIntervalNanoseconds: 1000, + DefaultMinObservationIntervalNanoseconds: 0, + } + require.NoError(t, cfg.Validate()) + }) + t.Run("DefaultMinObservationIntervalNanoseconds > DefaultMinReportIntervalNanoseconds is invalid", func(t *testing.T) { + cfg := OffchainConfig{ + ProtocolVersion: 1, + DefaultMinReportIntervalNanoseconds: 1000, + DefaultMinObservationIntervalNanoseconds: 1001, + } + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "default observation cadence must not exceed default report cadence") + }) }) t.Run("DefaultMinReportIntervalNanoseconds=0 is invalid", func(t *testing.T) { cfg := OffchainConfig{ diff --git a/llo/protocol/plugin_codecs.pb.go b/llo/protocol/plugin_codecs.pb.go index 159a4b2..a1d2dfd 100644 --- a/llo/protocol/plugin_codecs.pb.go +++ b/llo/protocol/plugin_codecs.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.12 -// protoc v7.35.1 +// protoc-gen-go v1.36.11 +// protoc v7.35.0 // source: plugin_codecs.proto package protocol @@ -1165,16 +1165,24 @@ func (x *LLOChannelStateProto) GetChannelDefinitions() []*LLOChannelIDAndDefinit // i.e. TimestampedStreamValues. Regular aggregates are recomputed fresh every // round and are never persisted. // +// observationDueNanoseconds is the observation schedule: the timestamp at which +// each channel next becomes due for observation and aggregation. It is distinct +// from validAfterNanoseconds, which is a report boundary emitted in the report +// itself. The schedule advances at a fixed rate from its own previous value, so +// a cycle that runs late does not push every later cycle out with it. A channel +// with no entry is due; entries appear once a channel has reported. +// // NOTE: must serialize deterministically, hence use of repeated tuple instead -// of maps. validAfterNanoseconds and reportableChannelIDs MUST be sorted -// ascending by channelID; streamAggregates MUST be sorted ascending by -// (streamID, aggregator). +// of maps. validAfterNanoseconds, reportableChannelIDs and +// observationDueNanoseconds MUST be sorted ascending by channelID; +// streamAggregates MUST be sorted ascending by (streamID, aggregator). type LLOHotStateProto struct { state protoimpl.MessageState `protogen:"open.v1"` ObservationTimestampNanoseconds uint64 `protobuf:"varint,1,opt,name=observationTimestampNanoseconds,proto3" json:"observationTimestampNanoseconds,omitempty"` ValidAfterNanoseconds []*LLOChannelIDAndValidAfterNanosecondsProto `protobuf:"bytes,2,rep,name=validAfterNanoseconds,proto3" json:"validAfterNanoseconds,omitempty"` ReportableChannelIDs []uint32 `protobuf:"varint,3,rep,packed,name=reportableChannelIDs,proto3" json:"reportableChannelIDs,omitempty"` StreamAggregates []*LLOStreamAggregate `protobuf:"bytes,4,rep,name=streamAggregates,proto3" json:"streamAggregates,omitempty"` + ObservationDueNanoseconds []*LLOChannelIDAndObservationDueProto `protobuf:"bytes,5,rep,name=observationDueNanoseconds,proto3" json:"observationDueNanoseconds,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1237,6 +1245,66 @@ func (x *LLOHotStateProto) GetStreamAggregates() []*LLOStreamAggregate { return nil } +func (x *LLOHotStateProto) GetObservationDueNanoseconds() []*LLOChannelIDAndObservationDueProto { + if x != nil { + return x.ObservationDueNanoseconds + } + return nil +} + +// LLOChannelIDAndObservationDueProto is one channel's observation schedule slot. +type LLOChannelIDAndObservationDueProto struct { + state protoimpl.MessageState `protogen:"open.v1"` + ChannelID uint32 `protobuf:"varint,1,opt,name=channelID,proto3" json:"channelID,omitempty"` + DueAtNanoseconds uint64 `protobuf:"varint,2,opt,name=dueAtNanoseconds,proto3" json:"dueAtNanoseconds,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LLOChannelIDAndObservationDueProto) Reset() { + *x = LLOChannelIDAndObservationDueProto{} + mi := &file_plugin_codecs_proto_msgTypes[18] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LLOChannelIDAndObservationDueProto) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LLOChannelIDAndObservationDueProto) ProtoMessage() {} + +func (x *LLOChannelIDAndObservationDueProto) ProtoReflect() protoreflect.Message { + mi := &file_plugin_codecs_proto_msgTypes[18] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LLOChannelIDAndObservationDueProto.ProtoReflect.Descriptor instead. +func (*LLOChannelIDAndObservationDueProto) Descriptor() ([]byte, []int) { + return file_plugin_codecs_proto_rawDescGZIP(), []int{18} +} + +func (x *LLOChannelIDAndObservationDueProto) GetChannelID() uint32 { + if x != nil { + return x.ChannelID + } + return 0 +} + +func (x *LLOChannelIDAndObservationDueProto) GetDueAtNanoseconds() uint64 { + if x != nil { + return x.DueAtNanoseconds + } + return 0 +} + // LLOPrecursorProto is the v31 ReportsPlusPrecursor: everything Reports needs, // since Reports gets no KeyValueStateReader. It mirrors LLOOutcomeProtoV1 (which // belongs to the v30 outcome and must not be changed for v31's benefit) and adds @@ -1261,7 +1329,7 @@ type LLOPrecursorProto struct { func (x *LLOPrecursorProto) Reset() { *x = LLOPrecursorProto{} - mi := &file_plugin_codecs_proto_msgTypes[18] + mi := &file_plugin_codecs_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1273,7 +1341,7 @@ func (x *LLOPrecursorProto) String() string { func (*LLOPrecursorProto) ProtoMessage() {} func (x *LLOPrecursorProto) ProtoReflect() protoreflect.Message { - mi := &file_plugin_codecs_proto_msgTypes[18] + mi := &file_plugin_codecs_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1286,7 +1354,7 @@ func (x *LLOPrecursorProto) ProtoReflect() protoreflect.Message { // Deprecated: Use LLOPrecursorProto.ProtoReflect.Descriptor instead. func (*LLOPrecursorProto) Descriptor() ([]byte, []int) { - return file_plugin_codecs_proto_rawDescGZIP(), []int{18} + return file_plugin_codecs_proto_rawDescGZIP(), []int{19} } func (x *LLOPrecursorProto) GetLifeCycleStage() string { @@ -1419,12 +1487,16 @@ const file_plugin_codecs_proto_rawDesc = "" + "aggregator\x18\x03 \x01(\rR\n" + "aggregator\"j\n" + "\x14LLOChannelStateProto\x12R\n" + - "\x12channelDefinitions\x18\x01 \x03(\v2\".v1.LLOChannelIDAndDefinitionProtoR\x12channelDefinitions\"\xb9\x02\n" + + "\x12channelDefinitions\x18\x01 \x03(\v2\".v1.LLOChannelIDAndDefinitionProtoR\x12channelDefinitions\"\x9f\x03\n" + "\x10LLOHotStateProto\x12H\n" + "\x1fobservationTimestampNanoseconds\x18\x01 \x01(\x04R\x1fobservationTimestampNanoseconds\x12c\n" + "\x15validAfterNanoseconds\x18\x02 \x03(\v2-.v1.LLOChannelIDAndValidAfterNanosecondsProtoR\x15validAfterNanoseconds\x122\n" + "\x14reportableChannelIDs\x18\x03 \x03(\rR\x14reportableChannelIDs\x12B\n" + - "\x10streamAggregates\x18\x04 \x03(\v2\x16.v1.LLOStreamAggregateR\x10streamAggregates\"\xb0\x03\n" + + "\x10streamAggregates\x18\x04 \x03(\v2\x16.v1.LLOStreamAggregateR\x10streamAggregates\x12d\n" + + "\x19observationDueNanoseconds\x18\x05 \x03(\v2&.v1.LLOChannelIDAndObservationDueProtoR\x19observationDueNanoseconds\"n\n" + + "\"LLOChannelIDAndObservationDueProto\x12\x1c\n" + + "\tchannelID\x18\x01 \x01(\rR\tchannelID\x12*\n" + + "\x10dueAtNanoseconds\x18\x02 \x01(\x04R\x10dueAtNanoseconds\"\xb0\x03\n" + "\x11LLOPrecursorProto\x12&\n" + "\x0elifeCycleStage\x18\x01 \x01(\tR\x0elifeCycleStage\x12H\n" + "\x1fobservationTimestampNanoseconds\x18\x02 \x01(\x04R\x1fobservationTimestampNanoseconds\x12R\n" + @@ -1447,7 +1519,7 @@ func file_plugin_codecs_proto_rawDescGZIP() []byte { } var file_plugin_codecs_proto_enumTypes = make([]protoimpl.EnumInfo, 1) -var file_plugin_codecs_proto_msgTypes = make([]protoimpl.MessageInfo, 21) +var file_plugin_codecs_proto_msgTypes = make([]protoimpl.MessageInfo, 22) var file_plugin_codecs_proto_goTypes = []any{ (LLOStreamValue_Type)(0), // 0: v1.LLOStreamValue.Type (*LLOObservationProto)(nil), // 1: v1.LLOObservationProto @@ -1468,13 +1540,14 @@ var file_plugin_codecs_proto_goTypes = []any{ (*LLOStreamAggregate)(nil), // 16: v1.LLOStreamAggregate (*LLOChannelStateProto)(nil), // 17: v1.LLOChannelStateProto (*LLOHotStateProto)(nil), // 18: v1.LLOHotStateProto - (*LLOPrecursorProto)(nil), // 19: v1.LLOPrecursorProto - nil, // 20: v1.LLOObservationProto.UpdateChannelDefinitionsEntry - nil, // 21: v1.LLOObservationProto.StreamValuesEntry + (*LLOChannelIDAndObservationDueProto)(nil), // 19: v1.LLOChannelIDAndObservationDueProto + (*LLOPrecursorProto)(nil), // 20: v1.LLOPrecursorProto + nil, // 21: v1.LLOObservationProto.UpdateChannelDefinitionsEntry + nil, // 22: v1.LLOObservationProto.StreamValuesEntry } var file_plugin_codecs_proto_depIdxs = []int32{ - 20, // 0: v1.LLOObservationProto.updateChannelDefinitions:type_name -> v1.LLOObservationProto.UpdateChannelDefinitionsEntry - 21, // 1: v1.LLOObservationProto.streamValues:type_name -> v1.LLOObservationProto.StreamValuesEntry + 21, // 0: v1.LLOObservationProto.updateChannelDefinitions:type_name -> v1.LLOObservationProto.UpdateChannelDefinitionsEntry + 22, // 1: v1.LLOObservationProto.streamValues:type_name -> v1.LLOObservationProto.StreamValuesEntry 0, // 2: v1.LLOStreamValue.type:type_name -> v1.LLOStreamValue.Type 2, // 3: v1.LLOTimestampedStreamValue.streamValue:type_name -> v1.LLOStreamValue 2, // 4: v1.LLOStreamHistoryRecord.value:type_name -> v1.LLOStreamValue @@ -1491,16 +1564,17 @@ var file_plugin_codecs_proto_depIdxs = []int32{ 13, // 15: v1.LLOChannelStateProto.channelDefinitions:type_name -> v1.LLOChannelIDAndDefinitionProto 15, // 16: v1.LLOHotStateProto.validAfterNanoseconds:type_name -> v1.LLOChannelIDAndValidAfterNanosecondsProto 16, // 17: v1.LLOHotStateProto.streamAggregates:type_name -> v1.LLOStreamAggregate - 13, // 18: v1.LLOPrecursorProto.channelDefinitions:type_name -> v1.LLOChannelIDAndDefinitionProto - 15, // 19: v1.LLOPrecursorProto.validAfterNanoseconds:type_name -> v1.LLOChannelIDAndValidAfterNanosecondsProto - 16, // 20: v1.LLOPrecursorProto.streamAggregates:type_name -> v1.LLOStreamAggregate - 8, // 21: v1.LLOObservationProto.UpdateChannelDefinitionsEntry.value:type_name -> v1.LLOChannelDefinitionProto - 2, // 22: v1.LLOObservationProto.StreamValuesEntry.value:type_name -> v1.LLOStreamValue - 23, // [23:23] is the sub-list for method output_type - 23, // [23:23] is the sub-list for method input_type - 23, // [23:23] is the sub-list for extension type_name - 23, // [23:23] is the sub-list for extension extendee - 0, // [0:23] is the sub-list for field type_name + 19, // 18: v1.LLOHotStateProto.observationDueNanoseconds:type_name -> v1.LLOChannelIDAndObservationDueProto + 13, // 19: v1.LLOPrecursorProto.channelDefinitions:type_name -> v1.LLOChannelIDAndDefinitionProto + 15, // 20: v1.LLOPrecursorProto.validAfterNanoseconds:type_name -> v1.LLOChannelIDAndValidAfterNanosecondsProto + 16, // 21: v1.LLOPrecursorProto.streamAggregates:type_name -> v1.LLOStreamAggregate + 8, // 22: v1.LLOObservationProto.UpdateChannelDefinitionsEntry.value:type_name -> v1.LLOChannelDefinitionProto + 2, // 23: v1.LLOObservationProto.StreamValuesEntry.value:type_name -> v1.LLOStreamValue + 24, // [24:24] is the sub-list for method output_type + 24, // [24:24] is the sub-list for method input_type + 24, // [24:24] is the sub-list for extension type_name + 24, // [24:24] is the sub-list for extension extendee + 0, // [0:24] is the sub-list for field type_name } func init() { file_plugin_codecs_proto_init() } @@ -1514,7 +1588,7 @@ func file_plugin_codecs_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_plugin_codecs_proto_rawDesc), len(file_plugin_codecs_proto_rawDesc)), NumEnums: 1, - NumMessages: 21, + NumMessages: 22, NumExtensions: 0, NumServices: 0, }, diff --git a/llo/protocol/plugin_codecs.proto b/llo/protocol/plugin_codecs.proto index 05aa6c4..5988fa3 100644 --- a/llo/protocol/plugin_codecs.proto +++ b/llo/protocol/plugin_codecs.proto @@ -183,15 +183,29 @@ message LLOChannelStateProto { // i.e. TimestampedStreamValues. Regular aggregates are recomputed fresh every // round and are never persisted. // +// observationDueNanoseconds is the observation schedule: the timestamp at which +// each channel next becomes due for observation and aggregation. It is distinct +// from validAfterNanoseconds, which is a report boundary emitted in the report +// itself. The schedule advances at a fixed rate from its own previous value, so +// a cycle that runs late does not push every later cycle out with it. A channel +// with no entry is due; entries appear once a channel has reported. +// // NOTE: must serialize deterministically, hence use of repeated tuple instead -// of maps. validAfterNanoseconds and reportableChannelIDs MUST be sorted -// ascending by channelID; streamAggregates MUST be sorted ascending by -// (streamID, aggregator). +// of maps. validAfterNanoseconds, reportableChannelIDs and +// observationDueNanoseconds MUST be sorted ascending by channelID; +// streamAggregates MUST be sorted ascending by (streamID, aggregator). message LLOHotStateProto { uint64 observationTimestampNanoseconds = 1; repeated LLOChannelIDAndValidAfterNanosecondsProto validAfterNanoseconds = 2; repeated uint32 reportableChannelIDs = 3; repeated LLOStreamAggregate streamAggregates = 4; + repeated LLOChannelIDAndObservationDueProto observationDueNanoseconds = 5; +} + +// LLOChannelIDAndObservationDueProto is one channel's observation schedule slot. +message LLOChannelIDAndObservationDueProto { + uint32 channelID = 1; + uint64 dueAtNanoseconds = 2; } // LLOPrecursorProto is the v31 ReportsPlusPrecursor: everything Reports needs,