feat(llo/v31): skip observation/aggregation for recently-reported channels - #263
Draft
cawthorne wants to merge 7 commits into
Draft
feat(llo/v31): skip observation/aggregation for recently-reported channels#263cawthorne wants to merge 7 commits into
cawthorne wants to merge 7 commits into
Conversation
…bservation/aggregation for recently-reported channels Add a new offchain config field that skips observation and aggregation for channels whose last report is too recent, mirroring the existing DefaultMinReportIntervalNanoseconds report-skip but applied earlier in the pipeline to avoid unnecessary work on channels that won't report this round. - New proto field 4: defaultMinObservationIntervalNanoseconds - Validation: must be 0 for protocol v0; must not exceed DefaultMinReportIntervalNanoseconds for v1+ (0 = disabled) - Observation: skips gathering stream values for not-due channels - Aggregation: skips aggregating streams exclusive to not-due channels - Calculated streams: only evaluated for due channels - Shared streams are still observed/aggregated via due channels - New channels are always due (no validAfter yet) - ValidAfter watermark advanced for reported channels before skip check, matching the one-round lag in persisted hot state
The blob pump serves a round from the snapshot gathered under the previous round's stream set, so a channel whose streams first entered the input on the round it became due was aggregated with no values at all. With the observation interval equal to the report interval the channel was due only on rounds it reported, so it was never due twice running and never recovered: every cycle produced a report with nil values (or a codec error) while validAfter advanced regardless, since isReportable does not require aggregates. Observation now applies the due check at now+lead, where lead is one round estimated from the gap to the previous round's agreed observation timestamp and capped at the interval. A channel's streams therefore enter the pump's input the round before the channel is due, and the snapshot the due round consumes already carries them. The lead also absorbs clock skew, so nodes whose clock trails the agreed median still observe what the median round will aggregate. Degenerate inputs (empty hot state, watermark ahead of now) widen the observed set rather than narrowing it. - SetInput now runs every round while Take stays gated on len(streams)>0, so the pump's input can never go stale and a round that needs no values no longer consumes and discards the snapshot gathered for the next one. - observableStreams exempts backfill channels from the due check instead of dropping them, restoring the observed set for hosts that have not configured the interval and matching the documented behaviour. - The due filter now lives in one place: StateTransition computes it once and feeds both aggregate and ProcessCalculatedStreams, and aggregate's signature reverts to its original shape. observableDefinitions retains tombstoned and backfill channels so the set handed to ProcessCalculatedStreams no longer depends on whether the interval is configured. - isObservationDue's contract is corrected: a missing watermark means due on the Observation side only. StateTransition seeds a watermark for every newly effective channel, so a new channel is not aggregated until the interval elapses, which costs it nothing because it was not reportable in that window.
…After Each channel now carries its own observation schedule in the hot state, advanced by the interval from its own previous slot and only on a round that reported. Deriving the skip from validAfter instead made the interval a fixed delay measured from whichever round actually reported, so latency in one cycle was added to every later cycle and the cadence crept. It also overloaded validAfter, which is a report boundary emitted in the report and defines the window (validAfter, observationTimestamp] that consecutive reports must tile exactly. With a schedule, latency in a cycle becomes a constant phase offset instead: a channel configured to report every T keeps reporting every T. That offset is also what gives the blob pump its lead. Values are gathered a round after a channel's streams enter the pump's input, so the first due round after a skip window has none and the channel withholds its report rather than emitting nils. Once schedule and watermark differ by that delay, a channel becomes due for observation far enough ahead of when it may report that its values are always ready, so the lead matches whatever the data source needs and is not configured anywhere. A channel with no schedule entry is due, so a newly effective one aggregates from its first round and builds its initial aggregates and history. It is still not reportable that round, since validAfter equals the observation timestamp. - LLOHotStateProto gains observationDueNanoseconds (field 5), sorted ascending by channelID like its neighbours for deterministic serialization - nextObservationDue owns the fixed-rate advance, and skips missed slots rather than firing every round to catch up after a channel has been unable to report - gofmt factory.go and plugin_test.go
A pair belonging only to a channel the schedule skipped was never visited by aggregate, so its carry-forward entry was dropped. Carry-forward is the last-known-good store behind TimestampedStreamValue monotonicity and behind surviving a transient aggregation failure, so dropping it let a channel adopt an older value coming out of a skip window than it held going in, and left it no fallback on the round it returned. Such pairs now keep their carried value. The pass is driven from the live definitions rather than from the carry map, so a pair whose last channel has been removed is still reclaimed by not being written into the new hot record, and it is restricted to pairs this round never visited, so the deliberate drops in the main loop keep their decision. Carried values are still kept out of history: appendHistory is not called for them and its strictly-newer guard would reject them anyway. A window must record a gap rather than a repeat, both so a carried value is not weighted twice and so that a stalled feed stays visible to TWAP's gap checks. Channels reading stream history are deliberately not exempt from the skip. The skip makes their report cadence the sampling rate, which lowers window resolution without making it wrong - records carry their own observation timestamp and TWAP integrates over real time - and it cannot stall silently, because 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. What that does change is how a window should be sized, so it is now documented where it will be read: on the config field and beside the warmup section in the calculated engine. TWAP buckets its window by the second and keeps the newest record per bucket, so an interval at or below a second leaves it unchanged and makes depth go further; above a second the observed bucket count falls to about window/interval. Every other window function reads the record series directly, so it tracks the sampling rate at any interval. A node also logs once per channel when a history-reading channel is first skipped.
Promotion replaces validAfter wholesale from the predecessor's retirement report so the handover is gapless, but the observation schedule was carried over from staging. A slot inherited that way can leave a channel not due on the promotion round; a channel that is not aggregated has no values, and so cannot report - reopening the gap up to a full interval wide, in the one path built to avoid it. An unset schedule means due, so clearing it lets every channel aggregate immediately and rebuild its slot from its first report, matching what validAfter already does.
The interval tests framed due-ness as validAfter+interval, which is how an earlier iteration derived it and is the conflation nextObservationDue warns against. The schedule is its own map and due-ness is now >= a stored slot, so name the map for what it is and state the comparisons the code actually makes. Rename readValidAfterOnly to readHotStateForObservation: it reads the schedule, observation timestamp and reportability flags, and Observation never reads the validAfter it was named for.
…terval A stream is gathered if any channel needing it is due, and aggregated once for every due channel declaring the same pair. Pairs wanted only by skipped channels are not aggregated, even when the underlying stream was gathered for someone else. Also pins the consequence that follows: a channel sharing a stream with a more frequently due one finds its values already gathered on the round it comes due, so it reports without the round it would otherwise withhold. Only a stream no due channel carries costs that round.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds an offchain config field
DefaultMinObservationIntervalNanoseconds. It is the same idea as the existingDefaultMinReportIntervalNanoseconds, moved earlier in the pipeline: instead of doing the work and discarding the report, a channel that cannot report yet is not observed or aggregated at all.Today a channel's streams are fetched (a data-source call, via the blob pump) and aggregated every round even when
DefaultMinReportIntervalNanosecondsguarantees it will not report. The saving scales with how much slower a channel's cadence is than the round rate.Unset (
0) everywhere today, and unset means no change: every channel is due and the observed set is byte-identical.How it works
Each channel carries an observation schedule in the hot state — the timestamp at which it next becomes due. The schedule advances by the interval from its own previous slot, and only on a round that actually reported.
That is fixed-rate rather than fixed-delay, and the distinction matters: advancing from the round that reported would add any latency in a cycle to every later cycle, so the cadence would creep. Advancing from the slot makes latency a one-time phase offset — a channel configured to report every
Tkeeps reporting everyT.The schedule 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.One consequence needs handling. Stream values are gathered asynchronously: they arrive a round after a channel's streams enter the pump's input. So on the first round after a skip window, a channel is due but its values have not landed yet. Reporting then would emit nils and advance
validAfteracross a round that carried nothing, so the channel instead withholds — it produces no report, and neither its watermark nor its schedule advances. It stays due and reports as soon as the values arrive.The one-off delay that creates becomes the standing offset between schedule and watermark, which is exactly the lead the pump needs: from then on a channel becomes due for observation far enough ahead of when it may report that its values are always ready. The lead is whatever the data source actually needs and is not configured anywhere.
Details worth a reviewer's attention
Shared streams. A stream is observed if any channel needing it is due, and aggregated once for every due channel declaring the same pair. Pairs wanted only by skipped channels are not aggregated. A channel sharing a stream with a more frequently due one therefore finds its values already gathered when it comes due, and skips the withheld round.
History sampling.
history_backfillchannels are exempt from the schedule — their watermark is a history timestamp, so a report cadence means nothing for them. Channels readingHistory(...)are not exempt, deliberately: the skip makes their report cadence the sampling rate for their windows. That lowers resolution without making results wrong (records carry their ownobservedAtNanoseconds, TWAP integrates over real time), and it cannot stall silently — an unreadable window leaves the channel unreportable, which stops its schedule advancing, so it reverts to observing every round. It does change how a window must be sized; the config field documents this per window function. A node logs once per channel the first time a history-reading channel is skipped.Carry-forward. A pair belonging only to a skipped channel keeps its carried value, so a channel does not come out of a skip window worse off than it went in. Carried values never reach history: a window records a gap rather than a repeat, so a stalled feed stays visible to TWAP's gap checks.
Promotion clears the schedule, matching how it replaces
validAfterwholesale from the predecessor's retirement report. A slot inherited from staging could otherwise leave a channel not due on the promotion round, reopening — up to a full interval wide — the very gap that handover exists to close.A channel with no schedule entry is due, so a newly effective channel aggregates from its first round and builds its initial history immediately.
Changes
llo/protocol/llo_offchain_config.protodefaultMinObservationIntervalNanosecondsllo/protocol/offchain_config.goDefaultMinReportIntervalNanoseconds(equal is fine)llo/protocol/plugin_codecs.protoLLOHotStateProtogainsobservationDueNanoseconds(field 5), sorted by channelID like its neighboursllo/dev/v31/plugin.goisObservationDue(schedule lookup) andnextObservationDue(fixed-rate advance, skipping missed slots rather than firing every round to catch up).Observationapplies the advance the one-round-lagging hot state has not recorded yetllo/dev/v31/statetransition.govalidAfter, and derives the due set once so aggregation and calculated-stream evaluation cannot disagreello/dev/v31/reports.gollo/dev/v31/kv.goreadHotStateForObservationreads what Observation needs without decoding carry-forward aggregatesTests
..._observableStreams..._SharedStreams,..._SharedStreamServesEveryDueChannel..._ObservationAdvancesLaggingScheduleTest_nextObservationDue..._WithholdsReportUntilValuesArrive..._CarryForwardSurvivesSkipTest_History_SurvivesObservationSkipTest_StateTransition_PromotionClearsObservationSchedule..._aggregate,..._FullRoundNotes for review
LLOHotStateProtois replicated state. The new field is additive, but writing it changes the serialized record, so rollout must be gated on the agreed offchain config like the rest of the feature.plugin_codecs.pb.gowas regenerated with protoc-gen-go v1.36.11 / protoc v7.35.0 against the committed v1.36.12 / v7.35.1, so the version header moves. Worth regenerating with the pinned toolchain before merge.