protocol

package
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	LifeCycleStageStaging    llotypes.LifeCycleStage = "staging"
	LifeCycleStageProduction llotypes.LifeCycleStage = "production"
	LifeCycleStageRetired    llotypes.LifeCycleStage = "retired"
)

Protocol instances start in either the staging or production stage. They may later be retired and "hand over" their work to another protocol instance that will move from the staging to the production stage.

These lifecycle constants and the retirement handover types are shared across all LLO plugin versions.

View Source 
const (
	// MaxReportCount is the maximum number of reports (and therefore channels)
	// supported. CAREFUL! If we ever accidentally exceed this e.g. through too
	// many channels/streams, the protocol will halt.
	// https://smartcontract-it.atlassian.net/browse/MERC-6468
	MaxReportCount = ocr3types.MaxMaxReportCount

	// Maximum amount of channels that can be removed per round (if more than
	// this need to be removed, they will be removed in batches until
	// everything is up-to-date)
	MaxObservationRemoveChannelIDsLength = 5
	// Maximum amount of channels that can be added/updated per round (if more
	// than this need to be added, they will be added in batches until
	// everything is up-to-date)
	MaxObservationUpdateChannelDefinitionsLength = 5
	// Maximum number of streams that can be observed per round
	MaxObservationStreamValuesLength = 10_000
	// Maximum allowed number of streams per channel
	MaxStreamsPerChannel = 10_000
	// MaxDecimalExponent bounds the absolute value of the base-10 exponent of
	// any decimal decoded from an untrusted source (peer observations, stored
	// state etc).
	// Stream values need only a couple of dozen decimal places, so this
	// leaves enough headroom.
	MaxDecimalExponent = 1_000
	// MaxOutcomeChannelDefinitionsLength is the maximum number of channels that
	// can be supported
	MaxOutcomeChannelDefinitionsLength = MaxReportCount

	// MaxHistoryRecordsPerPair bounds the depth of the persisted history window
	// for a single pair, and therefore the maximum depth any expression may
	// request. Note this is per pair, not per stream: a stream aggregated two
	// ways holds two windows of up to this depth each.
	MaxHistoryRecordsPerPair = 1024
	// MaxHistoryPairs bounds how many pairs may have history at once. Pairs are
	// ordered by (streamID, aggregator) and those beyond the cap are denied
	// history; channels referencing them become unreportable rather than
	// silently getting a shortened window. A stream aggregated two ways
	// consumes two of these, not one.
	MaxHistoryPairs = 128
	// MaxHistoryTotalBytes bounds the estimated total history bytes rewritten
	// per round (sum over pairs of requiredCount * MaxHistoryRecordBytes). The
	// OCR3.1 per-round budget for modified keys plus values is 10 MiB and is
	// shared with channel definitions and the other key prefixes, so history is
	// held well below it.
	MaxHistoryTotalBytes = 4 << 20
	// MaxHistoryRecordsPerExpression bounds the total history depth a single
	// expression may request, summed over all of its History calls. Each
	// requested record is work the evaluator does every round, so without this
	// one expression could combine many legal per-pair depths into an
	// arbitrarily expensive evaluation.
	MaxHistoryRecordsPerExpression = 4 * MaxHistoryRecordsPerPair

	// MaxTWAPCallsPerExpression bounds how many TWAP calls a single expression
	// may make.
	//
	// History depth is not a bound on TWAP work. A TWAP call needs only a
	// depth-1 history window, so MaxHistoryRecordsPerExpression permits
	// thousands of calls in one expression, and each is free to request the
	// longest window allowed — one-second buckets allocated and filled every
	// round, per channel, on the consensus path.
	//
	// Capping the count rather than pricing each call by its window is
	// deliberate: the count is syntactic, so it holds for a configuration built
	// at runtime, which a window-based budget could only bound by inspecting
	// literals. It is the coarser limit — every call is charged its worst case —
	// and that is the right trade for a limit consensus depends on.
	//
	// Four allows the shapes that need more than one window (a cross rate, a
	// spread between two TWAPs) while holding the per-round ceiling to four
	// maximum-length windows. Consensus-relevant, like every limit here: every
	// oracle must reject the same expression, so it is never per-node
	// configurable.
	MaxTWAPCallsPerExpression = 4
	// MaxHistoryRecordBytes is the maximum serialized size of one history
	// record, enforced on append (StreamHistory.Append) and used as the
	// per-record size when admitting pairs against MaxHistoryTotalBytes.
	// Enforcing the same number that the budget assumes is what makes the
	// budget a real bound rather than an estimate.
	//
	// It has to be enforced rather than assumed because MaxDecimalExponent
	// bounds a decimal's exponent but not its coefficient length: a 1000-digit
	// coefficient is ~415 bytes, so an unchecked Quote of three of them would be
	// ~1.3 KB per record and a full window ~1.3 MB — orders of magnitude past
	// what the byte budget was sized for, and still under libocr's 2 MiB
	// per-key limit, so nothing else would reject it.
	//
	// 128 B is roughly twice the largest measured record (a quote of three
	// 18-digit decimals, 60 B), so it accommodates real feeds while rejecting
	// pathological values. A rejected record leaves a gap in the series, which
	// is honest, and is logged.
	MaxHistoryRecordBytes = 128

	// MaxHistoryChunkRecords is the number of records held by one slot of the
	// chunked ring layout. It is the single knob of that layout: per-round write
	// bytes are proportional to it, per-round reads are proportional to depth
	// divided by it, and both are comfortable at 64.
	//
	// For a quote stream (60 B/record) a full chunk is 3.9 KiB, so a pair costs
	// ~1.9 KiB of writes in an average round instead of rewriting the whole
	// window, and a 1024-deep window is read in 18 point reads instead of 1024.
	// The value is consensus-relevant — it determines the bytes every oracle
	// writes — so it is a constant here and never per-node configurable.
	// Changing it changes the stored layout and requires resetting every window.
	//
	// It must divide MaxHistoryRecordsPerPair, so that a window at maximum depth
	// is an exact number of chunks (asserted by TestHistoryChunkLimits).
	MaxHistoryChunkRecords = 64
	// MaxHistoryChunkSlots is the size of the ring: the number of distinct chunk
	// slots one pair may occupy.
	//
	// A window retains at most MaxHistoryRecordsPerPair/MaxHistoryChunkRecords+1
	// chunks — the +1 for the partially consumed oldest chunk — and a round may
	// hold one more transiently, between appending into a freshly created chunk
	// and evicting the oldest. Hence +2.
	//
	// The ring is deliberately a fixed, small, statically known slot space
	// rather than an unbounded sequence: the in-round reader offers no range
	// scan, so if a header is ever unreadable there is no way to discover which
	// chunk keys exist. A bounded slot space makes recovery a blind delete of
	// every slot. Reuse of a slot across a lap of the ring is caught by the
	// sequence stored inside each chunk.
	MaxHistoryChunkSlots = MaxHistoryRecordsPerPair/MaxHistoryChunkRecords + 2
	// MaxHistoryRetainedRecords is the most records a window may hold. Retention
	// works in whole chunks, so a window overshoots its required depth by up to
	// one chunk less one record. Readers ask for an exact depth and never see
	// the overshoot.
	MaxHistoryRetainedRecords = MaxHistoryRecordsPerPair + MaxHistoryChunkRecords - 1
	// MaxHistoryHeaderBytes bounds the serialized size of a window header: a
	// couple of scalars plus two arrays of at most MaxHistoryChunkSlots entries,
	// which lands under 300 B. 512 leaves headroom without mattering to the
	// budget it feeds.
	MaxHistoryHeaderBytes = 512
	// MaxHistoryPairRoundBytes is what one pair can cost the per-round byte
	// budget: the newest chunk, rewritten every round, plus the header. Sealed
	// chunks are immutable and evictions are deletes, so nothing else is
	// written however deep the window is.
	//
	// This is the number that makes the chunked layout worth having: it is
	// independent of the window depth, where the single-blob layout charged
	// requiredCount * MaxHistoryRecordBytes every round.
	MaxHistoryPairRoundBytes = MaxHistoryChunkRecords*MaxHistoryRecordBytes + MaxHistoryHeaderBytes

	// MaxDecompressedObservationLength bounds the size of an observation after
	// zstd decompression.
	//
	// A legitimate observation is bounded by
	// MaxObservationStreamValuesLength stream values plus
	// MaxObservationUpdateChannelDefinitionsLength channel definitions of up
	// to MaxStreamsPerChannel streams each, which lands in the low single-digit
	// MiB range. 16 MiB leaves generous headroom.
	MaxDecompressedObservationLength = 16 << 20

	// MaxHistoryBackfillObservations bounds the maximum number of
	// observations to backfill per definition.
	MaxHistoryBackfillObservations = 64
)

Additional limits so we can more effectively bound the size of observations NOTE: These are hardcoded because these exact values are relied upon as a property of coming to consensus, it's too dangerous to make these configurable on a per-node basis. It may be possible to add them to the OffchainConfig if they need to be changed dynamically and in a backwards-compatible way.

These LLO-protocol limits are shared across all plugin versions.

View Source 
const (
	// DefaultMaxReportRange is the default maximum range of the report if unset in the opts.
	DefaultMaxReportRange = Duration(5 * time.Minute)
)

Variables

View Source 
var (
	LLOStreamValue_Type_name = map[int32]string{
		0: "Decimal",
		1: "Quote",
		2: "TimestampedStreamValue",
	}
	LLOStreamValue_Type_value = map[string]int32{
		"Decimal":                0,
		"Quote":                  1,
		"TimestampedStreamValue": 2,
	}
)

Enum value maps for LLOStreamValue_Type.

View Source 
var (
	// ErrCorruptStreamHistory is returned when a stored history window cannot
	// be decoded, or decodes into something that violates the type's
	// invariants (bad header, non-monotonic timestamps, over-capacity).
	//
	// Stored state is untrusted input: callers must handle this by discarding
	// the window and re-warming, never by panicking.
	ErrCorruptStreamHistory = errors.New("corrupt stream history")

	// ErrInsufficientStreamHistory is returned when fewer records are stored
	// than were asked for. It means "not yet evaluable" and must never be
	// treated as zero or as a shorter window.
	ErrInsufficientStreamHistory = errors.New("insufficient stream history")

	// ErrHistoryRecordTooLarge is returned when a value serializes to more than
	// MaxHistoryRecordBytes. Rejecting the record leaves an honest gap in the
	// series; accepting it would let one pair's window grow past what the
	// per-round byte budget was sized for.
	ErrHistoryRecordTooLarge = errors.New("history record too large")
)
View Source 
var (
	ErrNilStreamValue = errors.New("nil stream value")
	// ErrDecimalExponentOutOfRange is returned when a decimal decoded from an
	// untrusted source carries an exponent outside ±MaxDecimalExponent. A
	// well-behaved node never encodes such a value; accepting one would let a
	// single byzantine node force unbounded rescale work on every honest node.
	ErrDecimalExponentOutOfRange = errors.New("decimal exponent out of range")
)
View Source 
var ErrHistoryAlreadyAppended = errors.New("stream history already appended this round")

ErrHistoryAlreadyAppended is returned by RingWindow.Append when a pair is appended to twice in one round. See Append for why that cannot be allowed.

View Source 
var ErrHistoryChunkNotLoaded = errors.New("stream history chunk not loaded")

ErrHistoryChunkNotLoaded is returned when an operation needs a chunk the caller has not provided yet. It is a programming error, not corruption: the caller is expected to read exactly the chunks AppendPlan and ReadPlan name, and to hand them to Provide before mutating or reading the window.

View Source 
var File_attested_retirement_report_proto protoreflect.FileDescriptor
View Source 
var File_llo_offchain_config_proto protoreflect.FileDescriptor
View Source 
var File_llo_plugin_telemetry_proto protoreflect.FileDescriptor
View Source 
var File_plugin_codecs_proto protoreflect.FileDescriptor

Functions

func CalculatedStreamIDs added in v1.1.1 

func CalculatedStreamIDs(optsCache *OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) ([]llotypes.StreamID, error)

CalculatedStreamIDs returns the calculated stream IDs a channel's opts declare, in declaration order. It is the source of truth for which calculated streams a channel is expected to produce.

Returns an error if the opts cannot be resolved, declare no expressions, or declare a zero expression stream ID.

func ChangedChannelIDs added in v1.1.1 

func ChangedChannelIDs(current, desired llotypes.ChannelDefinitions) map[llotypes.ChannelID]struct{}

ChangedChannelIDs returns the IDs of the channels desired holds that current does not hold identically: the set being added or changed, which is the admitting set to verify against.

func ChannelDefinitionFromProto 

func ChannelDefinitionFromProto(pb *LLOChannelDefinitionProto) llotypes.ChannelDefinition

ChannelDefinitionFromProto decodes a protobuf ChannelDefinition. The caller must ensure pb is non-nil.

func CloneChannelDefinitions added in v1.1.1 

func CloneChannelDefinitions(in llotypes.ChannelDefinitions) llotypes.ChannelDefinitions

CloneChannelDefinitions returns a deep copy: the per-channel Streams slices and raw opts bytes are copied too, so appending to or otherwise mutating the clone cannot reach memory another round is reading.

func ConvertTimestamp 

func ConvertTimestamp(timestampNanos uint64, resolution TimeResolution) uint64

ConvertTimestamp converts a nanosecond timestamp to a specified resolution.

func Decode 

func Decode(value StreamValue, data []byte) error

func DropInvalidHistoryBackfillChannels 

func DropInvalidHistoryBackfillChannels(lggr logger.Logger, defs llotypes.ChannelDefinitions, nowNanos uint64) llotypes.ChannelDefinitions

DropInvalidHistoryBackfillChannels returns a copy of defs without history_backfill channels that fail validation. The input defs map is not modified. nowNanos should be wall-clock nanoseconds; use 0 to skip the future-timestamp check.

func EffectiveStreams added in v1.1.1 

func EffectiveStreams(optsCache *OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) ([]llotypes.Stream, error)

EffectiveStreams returns the streams a channel actually reports: the streams its definition observes, followed by one calculated stream per expression its opts declare, in declaration order.

This is a pure function of (definition, opts), which is the point. Calculated streams are derived, never stored: a definition is exactly what was voted on, and every node derives the same effective list from it without the derivation having to be replicated. Callers that need to know which streams a report carries — report value assembly above all — must go through here rather than reading cd.Streams directly.

The trailing position of the calculated streams is a contract, not an accident: ReportCodecEVMABIEncodeUnpackedExpr encodes the last len(opts.ABI) report values as its payload.

Inline llotypes.AggregatorCalculated entries on the definition are dropped before the derived ones are appended. Definitions written by older code carried the calculated streams inline, so filtering makes the result identical whether or not the stored definition was mutated, and makes the function idempotent under repeated application.

func GetOpts 

func GetOpts[T any](c *OptsCache, channelID llotypes.ChannelID) (T, error)

GetOpts returns decoded channel opts of type T for the given channel. On the first call for a given (channelID, T) after Set, the raw bytes are decoded via json.Unmarshal and the result is cached. Subsequent calls return the cached value directly.

Returns an error if the channel is not in the cache or decoding fails. The caller must pass a valid opts cache.

func HasCalculatedStreams added in v1.1.1 

func HasCalculatedStreams(cd llotypes.ChannelDefinition) bool

HasCalculatedStreams reports whether a channel definition's report format declares calculated streams. It is the definition's format, not its stream list, that answers this: calculated streams are derived from the opts and are not required to be present on the definition.

func HistoryChunkSlot added in v1.1.1 

func HistoryChunkSlot(sequence uint64) uint32

HistoryChunkSlot maps an absolute chunk sequence onto its ring slot.

func InitMemoryBallast 

func InitMemoryBallast()

InitMemoryBallast allocates the memory ballast, at most once per process. It is shared across plugin versions so that a process running both v30 and v31 holds a single ballast rather than one per version.

func ObservationTimestampKeyToNanoseconds 

func ObservationTimestampKeyToNanoseconds(rawKey uint64, res TimeResolution) (nanoseconds uint64, ok bool)

ObservationTimestampKeyToNanoseconds converts a raw observation timestamp key from opts to nanoseconds, reporting whether the key is representable.

The scaling is unsigned multiplication, so it wraps: a key beyond about 1.8e10 seconds becomes a small number of nanoseconds. That is the dangerous direction -- every caller compares the result against a bound it must be below (now, the round's observation timestamp) and a wrapped value passes all of them, so an unrepresentable far-future timestamp would read as a valid past one. Keys come from channel definition opts, which makes this reachable from configuration.

func ReportTimestampResolutionNanos 

func ReportTimestampResolutionNanos(target llotypes.ChannelDefinition) (uint64, error)

ReportTimestampResolutionNanos returns one tick of the target channel's observation timestamp resolution in nanoseconds.

func ScaleSeconds 

func ScaleSeconds(seconds uint32, resolution TimeResolution) uint64

ScaleSeconds converts a duration in seconds to a target resolution.

func SubtractChannelDefinitions 

func SubtractChannelDefinitions(minuend llotypes.ChannelDefinitions, subtrahend llotypes.ChannelDefinitions, limit int) llotypes.ChannelDefinitions

func ValidateHistoryBackfillAgainstDefinitions 

func ValidateHistoryBackfillAgainstDefinitions(cd llotypes.ChannelDefinition, defs llotypes.ChannelDefinitions, nowNanos uint64) error

ValidateHistoryBackfillAgainstDefinitions checks a single history_backfill definition against the full map. If nowNanos > 0, observation timestamps (converted to nanoseconds) must be < nowNanos.

func ValidateHistoryBackfillTarget added in v1.1.1 

func ValidateHistoryBackfillTarget(cd llotypes.ChannelDefinition, defs llotypes.ChannelDefinitions) error

ValidateHistoryBackfillTarget rejects a backfill channel whose target is itself a backfill channel, which includes a channel targeting itself.

Such a definition passes every other rule: the resolution lookup falls through to seconds, the stream lists match trivially when the target is the channel itself, and the target declares no calculated streams. It fails only at the very end, in Report, where the codec resolved from the target's report format is ReportCodecHistoryBackfill, whose Encode always errors. The channel is then selected, logged and skipped every round forever, because the watermark only advances on a report that was actually emitted.

This is an admission-only rule (see VerifyChannelDefinitionsForAdmission). Rejecting an already-committed definition here would stop an oracle from observing at all, and a definition like this has never produced a report, so there is nothing to protect but the ability to install a new one.

func VerifyChannelDefinitions 

func VerifyChannelDefinitions(codecs map[llotypes.ReportFormat]ReportCodec, channelDefs llotypes.ChannelDefinitions) error

VerifyChannelDefinitions applies the checks that any definition set must satisfy, whether it is being admitted or has already been committed.

func VerifyChannelDefinitionsForAdmission added in v1.1.1 

func VerifyChannelDefinitionsForAdmission(codecs map[llotypes.ReportFormat]ReportCodec, channelDefs llotypes.ChannelDefinitions, admitting map[llotypes.ChannelID]struct{}) error

VerifyChannelDefinitionsForAdmission additionally applies the admission-only checks, restricted to admitting -- the channels being added or changed.

The admission-only checks are the ones that reject a definition outright rather than merely stopping it from reporting: static expression analysis (ReportCodec implementations of AdmissionVerifier), calculated stream ID collisions, and feed ID uniqueness. Applying them to already-committed definitions would mean one grandfathered channel makes verification fail on every node, every round, which halts the protocol. Restricting them to admitting keeps the gate closed for anything new or changed while leaving what is already installed alone.

A cross-definition check involves two channels and is reported against whichever of them is seen second, so such a finding is kept when either channel is in admitting.

This is a local decision -- an oracle deciding what it is willing to vote for -- not a consensus-critical one, so oracles running different versions of the admission-only checks disagree only about what they vote for.

Types

type AdmissionVerifier added in v1.1.1 

type AdmissionVerifier interface {
	VerifyForAdmission(llotypes.ChannelDefinition) error
}

AdmissionVerifier is optionally implemented by a ReportCodec that has checks which must gate a definition being added or changed, but must not be applied to one that is already committed -- typically a check added after channels were already live, which a grandfathered definition may not satisfy.

VerifyForAdmission is only consulted for definitions Verify accepted, and only on the admission path; see VerifyChannelDefinitionsForAdmission. Like Verify it must be a pure function of the definition.

type AggregatorFunc 

type AggregatorFunc func(values []StreamValue, f int) (StreamValue, error)

func GetAggregatorFunc 

func GetAggregatorFunc(a llotypes.Aggregator) AggregatorFunc

type AttestedRetirementReport 

type AttestedRetirementReport struct {
	RetirementReport []byte                        `protobuf:"bytes,1,opt,name=retirementReport,proto3" json:"retirementReport,omitempty"`
	SeqNr            uint64                        `protobuf:"varint,2,opt,name=seqNr,proto3" json:"seqNr,omitempty"`
	Sigs             []*AttributedOnchainSignature `protobuf:"bytes,3,rep,name=sigs,proto3" json:"sigs,omitempty"`
	// contains filtered or unexported fields
}

func (*AttestedRetirementReport) Descriptor deprecated 

func (*AttestedRetirementReport) Descriptor() ([]byte, []int)

Deprecated: Use AttestedRetirementReport.ProtoReflect.Descriptor instead.

func (*AttestedRetirementReport) GetRetirementReport 

func (x *AttestedRetirementReport) GetRetirementReport() []byte

func (*AttestedRetirementReport) GetSeqNr 

func (x *AttestedRetirementReport) GetSeqNr() uint64

func (*AttestedRetirementReport) GetSigs 

func (x *AttestedRetirementReport) GetSigs() []*AttributedOnchainSignature

func (*AttestedRetirementReport) ProtoMessage 

func (*AttestedRetirementReport) ProtoMessage()

func (*AttestedRetirementReport) ProtoReflect 

func (x *AttestedRetirementReport) ProtoReflect() protoreflect.Message

func (*AttestedRetirementReport) Reset 

func (x *AttestedRetirementReport) Reset()

func (*AttestedRetirementReport) String 

func (x *AttestedRetirementReport) String() string

type AttributedOnchainSignature 

type AttributedOnchainSignature struct {
	Signature []byte `protobuf:"bytes,1,opt,name=signature,proto3" json:"signature,omitempty"`
	Signer    uint32 `protobuf:"varint,2,opt,name=signer,proto3" json:"signer,omitempty"`
	// contains filtered or unexported fields
}

func (*AttributedOnchainSignature) Descriptor deprecated 

func (*AttributedOnchainSignature) Descriptor() ([]byte, []int)

Deprecated: Use AttributedOnchainSignature.ProtoReflect.Descriptor instead.

func (*AttributedOnchainSignature) GetSignature 

func (x *AttributedOnchainSignature) GetSignature() []byte

func (*AttributedOnchainSignature) GetSigner 

func (x *AttributedOnchainSignature) GetSigner() uint32

func (*AttributedOnchainSignature) ProtoMessage 

func (*AttributedOnchainSignature) ProtoMessage()

func (*AttributedOnchainSignature) ProtoReflect 

func (x *AttributedOnchainSignature) ProtoReflect() protoreflect.Message

func (*AttributedOnchainSignature) Reset 

func (x *AttributedOnchainSignature) Reset()

func (*AttributedOnchainSignature) String 

func (x *AttributedOnchainSignature) String() string

type CalculatedStreamABI added in v1.1.1 

type CalculatedStreamABI struct {
	Type               string            `json:"type"`
	Expression         string            `json:"expression"`
	ExpressionStreamID llotypes.StreamID `json:"expressionStreamID"`
}

CalculatedStreamABI is one declared calculated stream: the expression that produces it and the stream ID it is published under.

type CalculatedStreamOpts added in v1.1.1 

type CalculatedStreamOpts struct {
	ABI []CalculatedStreamABI `json:"abi"`
}

CalculatedStreamOpts is the shape of the channel opts that declare calculated (expression) streams. It is the single decode of that shape: expression evaluation, channel definition verification and stream derivation all read it, so they cannot drift apart.

It lives here rather than in llo/protocol/calculated because that package depends on this one, and this package needs the shape in order to derive a channel's effective streams.

func DecodeCalculatedStreamOpts added in v1.1.1 

func DecodeCalculatedStreamOpts(optsCache *OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) (CalculatedStreamOpts, error)

DecodeCalculatedStreamOpts decodes a channel definition's calculated stream opts, preferring the (node-local) decode cache and falling back to decoding the definition's raw opts on a cache miss. The fallback keeps the result identical across oracles even when the cache has not been populated (e.g. after a restart, or in stages that never reset it).

Returns an error if the opts cannot be decoded or declare no expressions. The error does not name the channel: every caller already has the channel ID in hand and wraps with it, and naming it here produced it twice.

type ChannelCache added in v1.1.1 

type ChannelCache struct {
	// contains filtered or unexported fields
}

ChannelCache memoizes ChannelGenerations so that an unchanged record is read and decoded once rather than every round. It is purely a memo index: all consistency comes from generations being immutable and keyed by their record's sequence number.

The lookup is by equality, not "cached is older": a node replaying history or restoring from a snapshot can legitimately present an older record, and serving newer definitions into an older round would diverge.

A nil *ChannelCache is usable and simply memoizes nothing.

func NewChannelCache added in v1.1.1 

func NewChannelCache() *ChannelCache

func (*ChannelCache) Load added in v1.1.1 

func (c *ChannelCache) Load(seqNr uint64, build func() (llotypes.ChannelDefinitions, error)) (*ChannelGeneration, error)

Load returns the generation for seqNr, calling build only on a miss. build supplies the definitions of that record; they are deep-copied into the generation, so build may return memory it does not own exclusively (a decoded KV record, a precursor's definitions) without the generation aliasing it.

type ChannelDefinitionWithID 

type ChannelDefinitionWithID struct {
	llotypes.ChannelDefinition
	ChannelID llotypes.ChannelID
}

type ChannelGeneration added in v1.1.1 

type ChannelGeneration struct {
	// contains filtered or unexported fields
}

ChannelGeneration is an immutable snapshot of one channel-definitions record together with the decoded channel opts belonging to it. seqNr identifies the record (the v31 c/seqnr). The record is a pure function of that sequence number - both are written by the same StateTransition into the same replicated, atomically-committed store - so a generation is fully determined by its key.

A generation retains no reference to the store it was built from: no KeyValueState reader or transaction (those are invalid once the plugin callback returns), no back-pointer to the cache, and no memory shared with a mutable value. It is therefore safe to hold for the duration of a round and across goroutines, and - crucially - it can never be repointed at a different record by a concurrently running round. That is what keeps report encoding from using opts that are ahead of (or behind) the definitions being reported.

func (*ChannelGeneration) Definitions added in v1.1.1 

func (g *ChannelGeneration) Definitions() llotypes.ChannelDefinitions

Definitions returns the snapshot's channel definitions. They are read-only by contract: callers that mutate must clone first (see CloneChannelDefinitions).

func (*ChannelGeneration) Opts added in v1.1.1 

func (g *ChannelGeneration) Opts() *OptsCache

Opts returns the decoded-opts store for exactly these definitions. It is sealed: decode-only, and never re-synced to another record.

func (*ChannelGeneration) SeqNr added in v1.1.1 

func (g *ChannelGeneration) SeqNr() uint64

SeqNr returns the sequence number of the record this generation was built from.

type ChannelHash 

type ChannelHash [32]byte

type Decimal 

type Decimal decimal.Decimal

func ToDecimal 

func ToDecimal(d decimal.Decimal) *Decimal

func (*Decimal) Decimal 

func (v *Decimal) Decimal() decimal.Decimal

func (*Decimal) MarshalBinary 

func (v *Decimal) MarshalBinary() ([]byte, error)

func (*Decimal) MarshalText 

func (v *Decimal) MarshalText() ([]byte, error)

func (*Decimal) String 

func (v *Decimal) String() string

func (*Decimal) Type 

func (v *Decimal) Type() LLOStreamValue_Type

func (*Decimal) UnmarshalBinary 

func (v *Decimal) UnmarshalBinary(data []byte) error

func (*Decimal) UnmarshalText 

func (v *Decimal) UnmarshalText(data []byte) error

type Duration 

type Duration time.Duration

func (Duration) MarshalJSON 

func (d Duration) MarshalJSON() ([]byte, error)

func (Duration) String 

func (d Duration) String() string

func (*Duration) UnmarshalJSON 

func (d *Duration) UnmarshalJSON(b []byte) error

type EVMOnchainConfigCodec 

type EVMOnchainConfigCodec struct{}

EVMOnchainConfigCodec provides a llo-specific implementation of OnchainConfigCodec.

An encoded onchain config is expected to be in the format <version><predecessorConfigDigest> where version is a uint8 and min and max are in the format returned by EncodeValueInt192.

func (EVMOnchainConfigCodec) Decode 

func (EVMOnchainConfigCodec) Decode(b []byte) (OnchainConfig, error)

func (EVMOnchainConfigCodec) Encode 

func (EVMOnchainConfigCodec) Encode(c OnchainConfig) ([]byte, error)

type FeedIDer added in v1.1.1 

type FeedIDer interface {
	FeedID(llotypes.ChannelDefinition) (feedID [32]byte, ok bool, err error)
}

FeedIDer is optionally implemented by a ReportCodec whose reports are published under a feed ID. It exists so that VerifyChannelDefinitions can check feed IDs are unique across the whole definition set without knowing the format-specific opts the ID is carried in, or the format-specific rules for when a channel has one at all.

FeedID must be a pure function of the definition, like Verify, and is only consulted for definitions Verify accepted. The boolean is false for a channel whose reports carry no feed ID -- those are identified by channel ID, which is unique by construction and so needs no check.

type HistoryBackfillOpts 

type HistoryBackfillOpts struct {
	TargetChannelID llotypes.ChannelID `json:"targetChannelId"`
	// Observations maps raw timestamp keys (in the target channel's time resolution)
	// to stream ID -> serialized stream value string.
	Observations map[uint64]map[llotypes.StreamID]string `json:"-"`
}

HistoryBackfillOpts is the canonical JSON shape for history_backfill channel opts (after any DON-specific flattening into ChannelDefinition.Opts).

func GetHistoryBackfillOpts added in v1.1.1 

func GetHistoryBackfillOpts(optsCache *OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) (HistoryBackfillOpts, error)

GetHistoryBackfillOpts returns a history_backfill channel's parsed opts, preferring the (node-local) decode cache and falling back to parsing the definition's raw opts on a miss. The fallback keeps the result identical across oracles even when the cache has not been populated.

Selection runs two or three times per backfill channel per round, and the opts carry up to MaxHistoryBackfillObservations observations of stream values, so parsing them afresh each time is the difference between one allocation of that map per generation and several per round.

func ParseHistoryBackfillOpts 

func ParseHistoryBackfillOpts(raw llotypes.ChannelOpts) (HistoryBackfillOpts, error)

ParseHistoryBackfillOpts decodes opts bytes into HistoryBackfillOpts.

func (*HistoryBackfillOpts) UnmarshalJSON 

func (o *HistoryBackfillOpts) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes observations with string keys into uint64 maps.

type LLOAggregatorStreamValue 

type LLOAggregatorStreamValue struct {
	AggregatorValues map[uint32]*LLOStreamValue `` /* 184-byte string literal not displayed */
	// contains filtered or unexported fields
}

func (*LLOAggregatorStreamValue) Descriptor deprecated 

func (*LLOAggregatorStreamValue) Descriptor() ([]byte, []int)

Deprecated: Use LLOAggregatorStreamValue.ProtoReflect.Descriptor instead.

func (*LLOAggregatorStreamValue) GetAggregatorValues 

func (x *LLOAggregatorStreamValue) GetAggregatorValues() map[uint32]*LLOStreamValue

func (*LLOAggregatorStreamValue) ProtoMessage 

func (*LLOAggregatorStreamValue) ProtoMessage()

func (*LLOAggregatorStreamValue) ProtoReflect 

func (x *LLOAggregatorStreamValue) ProtoReflect() protoreflect.Message

func (*LLOAggregatorStreamValue) Reset 

func (x *LLOAggregatorStreamValue) Reset()

func (*LLOAggregatorStreamValue) String 

func (x *LLOAggregatorStreamValue) String() string

type LLOChannelDefinitionProto 

type LLOChannelDefinitionProto struct {
	ReportFormat           uint32                 `protobuf:"varint,1,opt,name=reportFormat,proto3" json:"reportFormat,omitempty"`
	Streams                []*LLOStreamDefinition `protobuf:"bytes,2,rep,name=streams,proto3" json:"streams,omitempty"`
	Opts                   []byte                 `protobuf:"bytes,3,opt,name=opts,proto3" json:"opts,omitempty"`
	Tombstone              bool                   `protobuf:"varint,4,opt,name=tombstone,proto3" json:"tombstone,omitempty"`
	Source                 uint32                 `protobuf:"varint,5,opt,name=source,proto3" json:"source,omitempty"`
	DisableNilStreamValues bool                   `protobuf:"varint,6,opt,name=disableNilStreamValues,proto3" json:"disableNilStreamValues,omitempty"`
	// contains filtered or unexported fields
}

func ChannelDefinitionToProto 

func ChannelDefinitionToProto(d llotypes.ChannelDefinition) *LLOChannelDefinitionProto

ChannelDefinitionToProto encodes a ChannelDefinition into its protobuf form.

func (*LLOChannelDefinitionProto) Descriptor deprecated 

func (*LLOChannelDefinitionProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOChannelDefinitionProto.ProtoReflect.Descriptor instead.

func (*LLOChannelDefinitionProto) GetDisableNilStreamValues 

func (x *LLOChannelDefinitionProto) GetDisableNilStreamValues() bool

func (*LLOChannelDefinitionProto) GetOpts 

func (x *LLOChannelDefinitionProto) GetOpts() []byte

func (*LLOChannelDefinitionProto) GetReportFormat 

func (x *LLOChannelDefinitionProto) GetReportFormat() uint32

func (*LLOChannelDefinitionProto) GetSource 

func (x *LLOChannelDefinitionProto) GetSource() uint32

func (*LLOChannelDefinitionProto) GetStreams 

func (x *LLOChannelDefinitionProto) GetStreams() []*LLOStreamDefinition

func (*LLOChannelDefinitionProto) GetTombstone 

func (x *LLOChannelDefinitionProto) GetTombstone() bool

func (*LLOChannelDefinitionProto) ProtoMessage 

func (*LLOChannelDefinitionProto) ProtoMessage()

func (*LLOChannelDefinitionProto) ProtoReflect 

func (x *LLOChannelDefinitionProto) ProtoReflect() protoreflect.Message

func (*LLOChannelDefinitionProto) Reset 

func (x *LLOChannelDefinitionProto) Reset()

func (*LLOChannelDefinitionProto) String 

func (x *LLOChannelDefinitionProto) String() string

type LLOChannelIDAndDefinitionProto 

type LLOChannelIDAndDefinitionProto struct {
	ChannelID         uint32                     `protobuf:"varint,1,opt,name=channelID,proto3" json:"channelID,omitempty"`
	ChannelDefinition *LLOChannelDefinitionProto `protobuf:"bytes,2,opt,name=channelDefinition,proto3" json:"channelDefinition,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOChannelIDAndDefinitionProto) Descriptor deprecated 

func (*LLOChannelIDAndDefinitionProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOChannelIDAndDefinitionProto.ProtoReflect.Descriptor instead.

func (*LLOChannelIDAndDefinitionProto) GetChannelDefinition 

func (x *LLOChannelIDAndDefinitionProto) GetChannelDefinition() *LLOChannelDefinitionProto

func (*LLOChannelIDAndDefinitionProto) GetChannelID 

func (x *LLOChannelIDAndDefinitionProto) GetChannelID() uint32

func (*LLOChannelIDAndDefinitionProto) ProtoMessage 

func (*LLOChannelIDAndDefinitionProto) ProtoMessage()

func (*LLOChannelIDAndDefinitionProto) ProtoReflect 

func (x *LLOChannelIDAndDefinitionProto) ProtoReflect() protoreflect.Message

func (*LLOChannelIDAndDefinitionProto) Reset 

func (x *LLOChannelIDAndDefinitionProto) Reset()

func (*LLOChannelIDAndDefinitionProto) String 

func (x *LLOChannelIDAndDefinitionProto) String() string

type LLOChannelIDAndValidAfterNanosecondsProto 

type LLOChannelIDAndValidAfterNanosecondsProto struct {
	ChannelID             uint32 `protobuf:"varint,1,opt,name=channelID,proto3" json:"channelID,omitempty"`
	ValidAfterNanoseconds uint64 `protobuf:"varint,2,opt,name=validAfterNanoseconds,proto3" json:"validAfterNanoseconds,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOChannelIDAndValidAfterNanosecondsProto) Descriptor deprecated 

func (*LLOChannelIDAndValidAfterNanosecondsProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOChannelIDAndValidAfterNanosecondsProto.ProtoReflect.Descriptor instead.

func (*LLOChannelIDAndValidAfterNanosecondsProto) GetChannelID 

func (x *LLOChannelIDAndValidAfterNanosecondsProto) GetChannelID() uint32

func (*LLOChannelIDAndValidAfterNanosecondsProto) GetValidAfterNanoseconds 

func (x *LLOChannelIDAndValidAfterNanosecondsProto) GetValidAfterNanoseconds() uint64

func (*LLOChannelIDAndValidAfterNanosecondsProto) ProtoMessage 

func (*LLOChannelIDAndValidAfterNanosecondsProto) ProtoMessage()

func (*LLOChannelIDAndValidAfterNanosecondsProto) ProtoReflect 

func (x *LLOChannelIDAndValidAfterNanosecondsProto) ProtoReflect() protoreflect.Message

func (*LLOChannelIDAndValidAfterNanosecondsProto) Reset 

func (x *LLOChannelIDAndValidAfterNanosecondsProto) Reset()

func (*LLOChannelIDAndValidAfterNanosecondsProto) String 

func (x *LLOChannelIDAndValidAfterNanosecondsProto) String() string

type LLOChannelIDAndValidAfterSecondsProto 

type LLOChannelIDAndValidAfterSecondsProto struct {
	ChannelID         uint32 `protobuf:"varint,1,opt,name=channelID,proto3" json:"channelID,omitempty"`
	ValidAfterSeconds uint32 `protobuf:"varint,2,opt,name=validAfterSeconds,proto3" json:"validAfterSeconds,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOChannelIDAndValidAfterSecondsProto) Descriptor deprecated 

func (*LLOChannelIDAndValidAfterSecondsProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOChannelIDAndValidAfterSecondsProto.ProtoReflect.Descriptor instead.

func (*LLOChannelIDAndValidAfterSecondsProto) GetChannelID 

func (x *LLOChannelIDAndValidAfterSecondsProto) GetChannelID() uint32

func (*LLOChannelIDAndValidAfterSecondsProto) GetValidAfterSeconds 

func (x *LLOChannelIDAndValidAfterSecondsProto) GetValidAfterSeconds() uint32

func (*LLOChannelIDAndValidAfterSecondsProto) ProtoMessage 

func (*LLOChannelIDAndValidAfterSecondsProto) ProtoMessage()

func (*LLOChannelIDAndValidAfterSecondsProto) ProtoReflect 

func (x *LLOChannelIDAndValidAfterSecondsProto) ProtoReflect() protoreflect.Message

func (*LLOChannelIDAndValidAfterSecondsProto) Reset 

func (x *LLOChannelIDAndValidAfterSecondsProto) Reset()

func (*LLOChannelIDAndValidAfterSecondsProto) String 

func (x *LLOChannelIDAndValidAfterSecondsProto) String() string

type LLOChannelStateProto added in v1.1.1 

type LLOChannelStateProto struct {
	ChannelDefinitions []*LLOChannelIDAndDefinitionProto `protobuf:"bytes,1,rep,name=channelDefinitions,proto3" json:"channelDefinitions,omitempty"`
	// contains filtered or unexported fields
}

LLOChannelStateProto is the v31 KeyValueState record holding the full set of live channel definitions under a single key (c/defs). It is rewritten only when the definitions change; c/seqnr records the sequence number of the last write so readers can cache it in memory across rounds.

NOTE: must serialize deterministically, hence use of repeated tuple instead of a map. channelDefinitions MUST be sorted ascending by channelID.

func (*LLOChannelStateProto) Descriptor deprecated added in v1.1.1 

func (*LLOChannelStateProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOChannelStateProto.ProtoReflect.Descriptor instead.

func (*LLOChannelStateProto) GetChannelDefinitions added in v1.1.1 

func (x *LLOChannelStateProto) GetChannelDefinitions() []*LLOChannelIDAndDefinitionProto

func (*LLOChannelStateProto) ProtoMessage added in v1.1.1 

func (*LLOChannelStateProto) ProtoMessage()

func (*LLOChannelStateProto) ProtoReflect added in v1.1.1 

func (x *LLOChannelStateProto) ProtoReflect() protoreflect.Message

func (*LLOChannelStateProto) Reset added in v1.1.1 

func (x *LLOChannelStateProto) Reset()

func (*LLOChannelStateProto) String added in v1.1.1 

func (x *LLOChannelStateProto) String() string

type LLOHotStateProto added in v1.1.1 

type LLOHotStateProto struct {
	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"`
	// contains filtered or unexported fields
}

LLOHotStateProto is the v31 KeyValueState record holding the per-round ("hot") state under a single key (r/agg): the state that changes on essentially every round.

streamAggregates carries only aggregates that must survive across rounds, i.e. TimestampedStreamValues. Regular aggregates are recomputed fresh every round and are never persisted.

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).

func (*LLOHotStateProto) Descriptor deprecated added in v1.1.1 

func (*LLOHotStateProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOHotStateProto.ProtoReflect.Descriptor instead.

func (*LLOHotStateProto) GetObservationTimestampNanoseconds added in v1.1.1 

func (x *LLOHotStateProto) GetObservationTimestampNanoseconds() uint64

func (*LLOHotStateProto) GetReportableChannelIDs added in v1.1.1 

func (x *LLOHotStateProto) GetReportableChannelIDs() []uint32

func (*LLOHotStateProto) GetStreamAggregates added in v1.1.1 

func (x *LLOHotStateProto) GetStreamAggregates() []*LLOStreamAggregate

func (*LLOHotStateProto) GetValidAfterNanoseconds added in v1.1.1 

func (x *LLOHotStateProto) GetValidAfterNanoseconds() []*LLOChannelIDAndValidAfterNanosecondsProto

func (*LLOHotStateProto) ProtoMessage added in v1.1.1 

func (*LLOHotStateProto) ProtoMessage()

func (*LLOHotStateProto) ProtoReflect added in v1.1.1 

func (x *LLOHotStateProto) ProtoReflect() protoreflect.Message

func (*LLOHotStateProto) Reset added in v1.1.1 

func (x *LLOHotStateProto) Reset()

func (*LLOHotStateProto) String added in v1.1.1 

func (x *LLOHotStateProto) String() string

type LLOObservationProto 

type LLOObservationProto struct {
	AttestedPredecessorRetirement []byte `protobuf:"bytes,1,opt,name=attestedPredecessorRetirement,proto3" json:"attestedPredecessorRetirement,omitempty"`
	ShouldRetire                  bool   `protobuf:"varint,2,opt,name=shouldRetire,proto3" json:"shouldRetire,omitempty"`
	// TODO: unixTimestampNanosecondsLegacy can be removed after this version
	// is rolled out everywhere
	UnixTimestampNanosecondsLegacy int64    `protobuf:"varint,3,opt,name=unixTimestampNanosecondsLegacy,proto3" json:"unixTimestampNanosecondsLegacy,omitempty"`
	UnixTimestampNanoseconds       uint64   `protobuf:"varint,7,opt,name=unixTimestampNanoseconds,proto3" json:"unixTimestampNanoseconds,omitempty"`
	RemoveChannelIDs               []uint32 `protobuf:"varint,4,rep,packed,name=removeChannelIDs,proto3" json:"removeChannelIDs,omitempty"`
	// Maps are safe to use here because Observation serialization does not
	// need to be deterministic. Non-deterministic map serialization is
	// marginally more efficient than converting to tuples and guarantees
	// uniqueness.
	UpdateChannelDefinitions map[uint32]*LLOChannelDefinitionProto `` /* 176-byte string literal not displayed */
	StreamValues             map[uint32]*LLOStreamValue            `` /* 152-byte string literal not displayed */
	// contains filtered or unexported fields
}

Observation CAN be changed as long as it doesn't break decode for legacy versions.

i.e. adding new fields is ok

func (*LLOObservationProto) Descriptor deprecated 

func (*LLOObservationProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOObservationProto.ProtoReflect.Descriptor instead.

func (*LLOObservationProto) GetAttestedPredecessorRetirement 

func (x *LLOObservationProto) GetAttestedPredecessorRetirement() []byte

func (*LLOObservationProto) GetRemoveChannelIDs 

func (x *LLOObservationProto) GetRemoveChannelIDs() []uint32

func (*LLOObservationProto) GetShouldRetire 

func (x *LLOObservationProto) GetShouldRetire() bool

func (*LLOObservationProto) GetStreamValues 

func (x *LLOObservationProto) GetStreamValues() map[uint32]*LLOStreamValue

func (*LLOObservationProto) GetUnixTimestampNanoseconds 

func (x *LLOObservationProto) GetUnixTimestampNanoseconds() uint64

func (*LLOObservationProto) GetUnixTimestampNanosecondsLegacy 

func (x *LLOObservationProto) GetUnixTimestampNanosecondsLegacy() int64

func (*LLOObservationProto) GetUpdateChannelDefinitions 

func (x *LLOObservationProto) GetUpdateChannelDefinitions() map[uint32]*LLOChannelDefinitionProto

func (*LLOObservationProto) ProtoMessage 

func (*LLOObservationProto) ProtoMessage()

func (*LLOObservationProto) ProtoReflect 

func (x *LLOObservationProto) ProtoReflect() protoreflect.Message

func (*LLOObservationProto) Reset 

func (x *LLOObservationProto) Reset()

func (*LLOObservationProto) String 

func (x *LLOObservationProto) String() string

type LLOOffchainConfigProto 

type LLOOffchainConfigProto struct {
	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"`
	// contains filtered or unexported fields
}

func (*LLOOffchainConfigProto) Descriptor deprecated 

func (*LLOOffchainConfigProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOOffchainConfigProto.ProtoReflect.Descriptor instead.

func (*LLOOffchainConfigProto) GetDefaultMinReportIntervalNanoseconds 

func (x *LLOOffchainConfigProto) GetDefaultMinReportIntervalNanoseconds() uint64

func (*LLOOffchainConfigProto) GetEnableObservationCompression 

func (x *LLOOffchainConfigProto) GetEnableObservationCompression() bool

func (*LLOOffchainConfigProto) GetProtocolVersion 

func (x *LLOOffchainConfigProto) GetProtocolVersion() uint32

func (*LLOOffchainConfigProto) ProtoMessage 

func (*LLOOffchainConfigProto) ProtoMessage()

func (*LLOOffchainConfigProto) ProtoReflect 

func (x *LLOOffchainConfigProto) ProtoReflect() protoreflect.Message

func (*LLOOffchainConfigProto) Reset 

func (x *LLOOffchainConfigProto) Reset()

func (*LLOOffchainConfigProto) String 

func (x *LLOOffchainConfigProto) String() string

type LLOOutcomeProtoV0 

type LLOOutcomeProtoV0 struct {
	LifeCycleStage                  string                                   `protobuf:"bytes,1,opt,name=lifeCycleStage,proto3" json:"lifeCycleStage,omitempty"`
	ObservationTimestampNanoseconds int64                                    `protobuf:"varint,2,opt,name=observationTimestampNanoseconds,proto3" json:"observationTimestampNanoseconds,omitempty"`
	ChannelDefinitions              []*LLOChannelIDAndDefinitionProto        `protobuf:"bytes,3,rep,name=channelDefinitions,proto3" json:"channelDefinitions,omitempty"`
	ValidAfterSeconds               []*LLOChannelIDAndValidAfterSecondsProto `protobuf:"bytes,4,rep,name=validAfterSeconds,proto3" json:"validAfterSeconds,omitempty"`
	StreamAggregates                []*LLOStreamAggregate                    `protobuf:"bytes,5,rep,name=streamAggregates,proto3" json:"streamAggregates,omitempty"`
	// contains filtered or unexported fields
}

NOTE: Outcome must serialize deterministically, hence use of repeated tuple instead of maps

func (*LLOOutcomeProtoV0) Descriptor deprecated 

func (*LLOOutcomeProtoV0) Descriptor() ([]byte, []int)

Deprecated: Use LLOOutcomeProtoV0.ProtoReflect.Descriptor instead.

func (*LLOOutcomeProtoV0) GetChannelDefinitions 

func (x *LLOOutcomeProtoV0) GetChannelDefinitions() []*LLOChannelIDAndDefinitionProto

func (*LLOOutcomeProtoV0) GetLifeCycleStage 

func (x *LLOOutcomeProtoV0) GetLifeCycleStage() string

func (*LLOOutcomeProtoV0) GetObservationTimestampNanoseconds 

func (x *LLOOutcomeProtoV0) GetObservationTimestampNanoseconds() int64

func (*LLOOutcomeProtoV0) GetStreamAggregates 

func (x *LLOOutcomeProtoV0) GetStreamAggregates() []*LLOStreamAggregate

func (*LLOOutcomeProtoV0) GetValidAfterSeconds 

func (x *LLOOutcomeProtoV0) GetValidAfterSeconds() []*LLOChannelIDAndValidAfterSecondsProto

func (*LLOOutcomeProtoV0) ProtoMessage 

func (*LLOOutcomeProtoV0) ProtoMessage()

func (*LLOOutcomeProtoV0) ProtoReflect 

func (x *LLOOutcomeProtoV0) ProtoReflect() protoreflect.Message

func (*LLOOutcomeProtoV0) Reset 

func (x *LLOOutcomeProtoV0) Reset()

func (*LLOOutcomeProtoV0) String 

func (x *LLOOutcomeProtoV0) String() string

type LLOOutcomeProtoV1 

type LLOOutcomeProtoV1 struct {
	LifeCycleStage                  string                                       `protobuf:"bytes,1,opt,name=lifeCycleStage,proto3" json:"lifeCycleStage,omitempty"`
	ObservationTimestampNanoseconds uint64                                       `protobuf:"varint,2,opt,name=observationTimestampNanoseconds,proto3" json:"observationTimestampNanoseconds,omitempty"`
	ChannelDefinitions              []*LLOChannelIDAndDefinitionProto            `protobuf:"bytes,3,rep,name=channelDefinitions,proto3" json:"channelDefinitions,omitempty"`
	ValidAfterNanoseconds           []*LLOChannelIDAndValidAfterNanosecondsProto `protobuf:"bytes,4,rep,name=validAfterNanoseconds,proto3" json:"validAfterNanoseconds,omitempty"`
	StreamAggregates                []*LLOStreamAggregate                        `protobuf:"bytes,5,rep,name=streamAggregates,proto3" json:"streamAggregates,omitempty"`
	// contains filtered or unexported fields
}

NOTE: Outcome must serialize deterministically, hence use of repeated tuple instead of maps

func (*LLOOutcomeProtoV1) Descriptor deprecated 

func (*LLOOutcomeProtoV1) Descriptor() ([]byte, []int)

Deprecated: Use LLOOutcomeProtoV1.ProtoReflect.Descriptor instead.

func (*LLOOutcomeProtoV1) GetChannelDefinitions 

func (x *LLOOutcomeProtoV1) GetChannelDefinitions() []*LLOChannelIDAndDefinitionProto

func (*LLOOutcomeProtoV1) GetLifeCycleStage 

func (x *LLOOutcomeProtoV1) GetLifeCycleStage() string

func (*LLOOutcomeProtoV1) GetObservationTimestampNanoseconds 

func (x *LLOOutcomeProtoV1) GetObservationTimestampNanoseconds() uint64

func (*LLOOutcomeProtoV1) GetStreamAggregates 

func (x *LLOOutcomeProtoV1) GetStreamAggregates() []*LLOStreamAggregate

func (*LLOOutcomeProtoV1) GetValidAfterNanoseconds 

func (x *LLOOutcomeProtoV1) GetValidAfterNanoseconds() []*LLOChannelIDAndValidAfterNanosecondsProto

func (*LLOOutcomeProtoV1) ProtoMessage 

func (*LLOOutcomeProtoV1) ProtoMessage()

func (*LLOOutcomeProtoV1) ProtoReflect 

func (x *LLOOutcomeProtoV1) ProtoReflect() protoreflect.Message

func (*LLOOutcomeProtoV1) Reset 

func (x *LLOOutcomeProtoV1) Reset()

func (*LLOOutcomeProtoV1) String 

func (x *LLOOutcomeProtoV1) String() string

type LLOOutcomeTelemetry 

type LLOOutcomeTelemetry struct {
	LifeCycleStage                  string `protobuf:"bytes,1,opt,name=life_cycle_stage,json=lifeCycleStage,proto3" json:"life_cycle_stage,omitempty"`
	ObservationTimestampNanoseconds uint64 `` /* 157-byte string literal not displayed */
	// channel id => channel definition
	ChannelDefinitions map[uint32]*LLOChannelDefinitionProto `` /* 190-byte string literal not displayed */
	// channel id => valid after nanoseconds
	ValidAfterNanoseconds map[uint32]uint64 `` /* 202-byte string literal not displayed */
	// stream id => aggregator => value
	StreamAggregates map[uint32]*LLOAggregatorStreamValue `` /* 184-byte string literal not displayed */
	SeqNr            uint64                               `protobuf:"varint,9,opt,name=seq_nr,json=seqNr,proto3" json:"seq_nr,omitempty"`
	ConfigDigest     []byte                               `protobuf:"bytes,10,opt,name=config_digest,json=configDigest,proto3" json:"config_digest,omitempty"`
	DonId            uint32                               `protobuf:"varint,11,opt,name=don_id,json=donId,proto3" json:"don_id,omitempty"`
	// contains filtered or unexported fields
}

LLOOutcomeTelemetry sent on every call to Outcome (once per round)

func (*LLOOutcomeTelemetry) Descriptor deprecated 

func (*LLOOutcomeTelemetry) Descriptor() ([]byte, []int)

Deprecated: Use LLOOutcomeTelemetry.ProtoReflect.Descriptor instead.

func (*LLOOutcomeTelemetry) GetChannelDefinitions 

func (x *LLOOutcomeTelemetry) GetChannelDefinitions() map[uint32]*LLOChannelDefinitionProto

func (*LLOOutcomeTelemetry) GetConfigDigest 

func (x *LLOOutcomeTelemetry) GetConfigDigest() []byte

func (*LLOOutcomeTelemetry) GetDonId 

func (x *LLOOutcomeTelemetry) GetDonId() uint32

func (*LLOOutcomeTelemetry) GetLifeCycleStage 

func (x *LLOOutcomeTelemetry) GetLifeCycleStage() string

func (*LLOOutcomeTelemetry) GetObservationTimestampNanoseconds 

func (x *LLOOutcomeTelemetry) GetObservationTimestampNanoseconds() uint64

func (*LLOOutcomeTelemetry) GetSeqNr 

func (x *LLOOutcomeTelemetry) GetSeqNr() uint64

func (*LLOOutcomeTelemetry) GetStreamAggregates 

func (x *LLOOutcomeTelemetry) GetStreamAggregates() map[uint32]*LLOAggregatorStreamValue

func (*LLOOutcomeTelemetry) GetValidAfterNanoseconds 

func (x *LLOOutcomeTelemetry) GetValidAfterNanoseconds() map[uint32]uint64

func (*LLOOutcomeTelemetry) ProtoMessage 

func (*LLOOutcomeTelemetry) ProtoMessage()

func (*LLOOutcomeTelemetry) ProtoReflect 

func (x *LLOOutcomeTelemetry) ProtoReflect() protoreflect.Message

func (*LLOOutcomeTelemetry) Reset 

func (x *LLOOutcomeTelemetry) Reset()

func (*LLOOutcomeTelemetry) String 

func (x *LLOOutcomeTelemetry) String() string

type LLOPrecursorProto added in v1.1.1 

type LLOPrecursorProto struct {
	LifeCycleStage                  string                                       `protobuf:"bytes,1,opt,name=lifeCycleStage,proto3" json:"lifeCycleStage,omitempty"`
	ObservationTimestampNanoseconds uint64                                       `protobuf:"varint,2,opt,name=observationTimestampNanoseconds,proto3" json:"observationTimestampNanoseconds,omitempty"`
	ChannelDefinitions              []*LLOChannelIDAndDefinitionProto            `protobuf:"bytes,3,rep,name=channelDefinitions,proto3" json:"channelDefinitions,omitempty"`
	ValidAfterNanoseconds           []*LLOChannelIDAndValidAfterNanosecondsProto `protobuf:"bytes,4,rep,name=validAfterNanoseconds,proto3" json:"validAfterNanoseconds,omitempty"`
	StreamAggregates                []*LLOStreamAggregate                        `protobuf:"bytes,5,rep,name=streamAggregates,proto3" json:"streamAggregates,omitempty"`
	ChannelStateSeqNr               uint64                                       `protobuf:"varint,6,opt,name=channelStateSeqNr,proto3" json:"channelStateSeqNr,omitempty"`
	// contains filtered or unexported fields
}

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 the sequence number of the channel-definitions record the precursor was built from, so that consumers can tell whether their decoded-opts cache already matches these definitions.

NOTE: must serialize deterministically, hence use of repeated tuple instead of maps. channelDefinitions and validAfterNanoseconds MUST be sorted ascending by channelID; streamAggregates MUST be sorted ascending by (streamID, aggregator).

func (*LLOPrecursorProto) Descriptor deprecated added in v1.1.1 

func (*LLOPrecursorProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOPrecursorProto.ProtoReflect.Descriptor instead.

func (*LLOPrecursorProto) GetChannelDefinitions added in v1.1.1 

func (x *LLOPrecursorProto) GetChannelDefinitions() []*LLOChannelIDAndDefinitionProto

func (*LLOPrecursorProto) GetChannelStateSeqNr added in v1.1.1 

func (x *LLOPrecursorProto) GetChannelStateSeqNr() uint64

func (*LLOPrecursorProto) GetLifeCycleStage added in v1.1.1 

func (x *LLOPrecursorProto) GetLifeCycleStage() string

func (*LLOPrecursorProto) GetObservationTimestampNanoseconds added in v1.1.1 

func (x *LLOPrecursorProto) GetObservationTimestampNanoseconds() uint64

func (*LLOPrecursorProto) GetStreamAggregates added in v1.1.1 

func (x *LLOPrecursorProto) GetStreamAggregates() []*LLOStreamAggregate

func (*LLOPrecursorProto) GetValidAfterNanoseconds added in v1.1.1 

func (x *LLOPrecursorProto) GetValidAfterNanoseconds() []*LLOChannelIDAndValidAfterNanosecondsProto

func (*LLOPrecursorProto) ProtoMessage added in v1.1.1 

func (*LLOPrecursorProto) ProtoMessage()

func (*LLOPrecursorProto) ProtoReflect added in v1.1.1 

func (x *LLOPrecursorProto) ProtoReflect() protoreflect.Message

func (*LLOPrecursorProto) Reset added in v1.1.1 

func (x *LLOPrecursorProto) Reset()

func (*LLOPrecursorProto) String added in v1.1.1 

func (x *LLOPrecursorProto) String() string

type LLOReportTelemetry 

type LLOReportTelemetry struct {
	ChannelId                       uint32                 `protobuf:"varint,1,opt,name=channel_id,json=channelId,proto3" json:"channel_id,omitempty"`
	ValidAfterNanoseconds           uint64                 `` /* 127-byte string literal not displayed */
	ObservationTimestampNanoseconds uint64                 `` /* 157-byte string literal not displayed */
	ReportFormat                    uint32                 `protobuf:"varint,4,opt,name=report_format,json=reportFormat,proto3" json:"report_format,omitempty"`
	Specimen                        bool                   `protobuf:"varint,5,opt,name=specimen,proto3" json:"specimen,omitempty"`
	StreamDefinitions               []*LLOStreamDefinition `protobuf:"bytes,6,rep,name=stream_definitions,json=streamDefinitions,proto3" json:"stream_definitions,omitempty"`
	StreamValues                    []*LLOStreamValue      `protobuf:"bytes,7,rep,name=stream_values,json=streamValues,proto3" json:"stream_values,omitempty"`
	ChannelOpts                     []byte                 `protobuf:"bytes,8,opt,name=channel_opts,json=channelOpts,proto3" json:"channel_opts,omitempty"`
	SeqNr                           uint64                 `protobuf:"varint,9,opt,name=seq_nr,json=seqNr,proto3" json:"seq_nr,omitempty"`
	ConfigDigest                    []byte                 `protobuf:"bytes,10,opt,name=config_digest,json=configDigest,proto3" json:"config_digest,omitempty"`
	DonId                           uint32                 `protobuf:"varint,11,opt,name=don_id,json=donId,proto3" json:"don_id,omitempty"`
	// contains filtered or unexported fields
}

LLOReportTelemetry sent for each report on every call to Reports

func (*LLOReportTelemetry) Descriptor deprecated 

func (*LLOReportTelemetry) Descriptor() ([]byte, []int)

Deprecated: Use LLOReportTelemetry.ProtoReflect.Descriptor instead.

func (*LLOReportTelemetry) GetChannelId 

func (x *LLOReportTelemetry) GetChannelId() uint32

func (*LLOReportTelemetry) GetChannelOpts 

func (x *LLOReportTelemetry) GetChannelOpts() []byte

func (*LLOReportTelemetry) GetConfigDigest 

func (x *LLOReportTelemetry) GetConfigDigest() []byte

func (*LLOReportTelemetry) GetDonId 

func (x *LLOReportTelemetry) GetDonId() uint32

func (*LLOReportTelemetry) GetObservationTimestampNanoseconds 

func (x *LLOReportTelemetry) GetObservationTimestampNanoseconds() uint64

func (*LLOReportTelemetry) GetReportFormat 

func (x *LLOReportTelemetry) GetReportFormat() uint32

func (*LLOReportTelemetry) GetSeqNr 

func (x *LLOReportTelemetry) GetSeqNr() uint64

func (*LLOReportTelemetry) GetSpecimen 

func (x *LLOReportTelemetry) GetSpecimen() bool

func (*LLOReportTelemetry) GetStreamDefinitions 

func (x *LLOReportTelemetry) GetStreamDefinitions() []*LLOStreamDefinition

func (*LLOReportTelemetry) GetStreamValues 

func (x *LLOReportTelemetry) GetStreamValues() []*LLOStreamValue

func (*LLOReportTelemetry) GetValidAfterNanoseconds 

func (x *LLOReportTelemetry) GetValidAfterNanoseconds() uint64

func (*LLOReportTelemetry) ProtoMessage 

func (*LLOReportTelemetry) ProtoMessage()

func (*LLOReportTelemetry) ProtoReflect 

func (x *LLOReportTelemetry) ProtoReflect() protoreflect.Message

func (*LLOReportTelemetry) Reset 

func (x *LLOReportTelemetry) Reset()

func (*LLOReportTelemetry) String 

func (x *LLOReportTelemetry) String() string

type LLOStreamAggregate 

type LLOStreamAggregate struct {
	StreamID    uint32          `protobuf:"varint,1,opt,name=streamID,proto3" json:"streamID,omitempty"`
	StreamValue *LLOStreamValue `protobuf:"bytes,2,opt,name=streamValue,proto3" json:"streamValue,omitempty"`
	Aggregator  uint32          `protobuf:"varint,3,opt,name=aggregator,proto3" json:"aggregator,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOStreamAggregate) Descriptor deprecated 

func (*LLOStreamAggregate) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamAggregate.ProtoReflect.Descriptor instead.

func (*LLOStreamAggregate) GetAggregator 

func (x *LLOStreamAggregate) GetAggregator() uint32

func (*LLOStreamAggregate) GetStreamID 

func (x *LLOStreamAggregate) GetStreamID() uint32

func (*LLOStreamAggregate) GetStreamValue 

func (x *LLOStreamAggregate) GetStreamValue() *LLOStreamValue

func (*LLOStreamAggregate) ProtoMessage 

func (*LLOStreamAggregate) ProtoMessage()

func (*LLOStreamAggregate) ProtoReflect 

func (x *LLOStreamAggregate) ProtoReflect() protoreflect.Message

func (*LLOStreamAggregate) Reset 

func (x *LLOStreamAggregate) Reset()

func (*LLOStreamAggregate) String 

func (x *LLOStreamAggregate) String() string

type LLOStreamDefinition 

type LLOStreamDefinition struct {
	StreamID   uint32 `protobuf:"varint,1,opt,name=streamID,proto3" json:"streamID,omitempty"`
	Aggregator uint32 `protobuf:"varint,2,opt,name=aggregator,proto3" json:"aggregator,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOStreamDefinition) Descriptor deprecated 

func (*LLOStreamDefinition) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamDefinition.ProtoReflect.Descriptor instead.

func (*LLOStreamDefinition) GetAggregator 

func (x *LLOStreamDefinition) GetAggregator() uint32

func (*LLOStreamDefinition) GetStreamID 

func (x *LLOStreamDefinition) GetStreamID() uint32

func (*LLOStreamDefinition) ProtoMessage 

func (*LLOStreamDefinition) ProtoMessage()

func (*LLOStreamDefinition) ProtoReflect 

func (x *LLOStreamDefinition) ProtoReflect() protoreflect.Message

func (*LLOStreamDefinition) Reset 

func (x *LLOStreamDefinition) Reset()

func (*LLOStreamDefinition) String 

func (x *LLOStreamDefinition) String() string

type LLOStreamHistoryChunkProto added in v1.1.1 

type LLOStreamHistoryChunkProto struct {

	// Absolute chunk index, monotonically increasing over the life of the
	// window. The key holds only sequence mod MaxHistoryChunkSlots, so this is
	// what distinguishes a live chunk from a stale one left by an earlier lap of
	// the ring.
	Sequence uint64 `protobuf:"varint,1,opt,name=sequence,proto3" json:"sequence,omitempty"`
	// Oldest first, newest last; observedAtNanoseconds strictly increasing.
	Records []*LLOStreamHistoryRecord `protobuf:"bytes,2,rep,name=records,proto3" json:"records,omitempty"`
	// contains filtered or unexported fields
}

LLOStreamHistoryChunkProto is one slot of the chunked ring layout: a run of consecutive history records for a single (streamID, aggregator) pair.

Only the newest chunk of a window is ever rewritten; once a chunk is full it is immutable for as long as it is retained. That is what makes the per-round write cost a function of the chunk size rather than of the window depth.

func (*LLOStreamHistoryChunkProto) Descriptor deprecated added in v1.1.1 

func (*LLOStreamHistoryChunkProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamHistoryChunkProto.ProtoReflect.Descriptor instead.

func (*LLOStreamHistoryChunkProto) GetRecords added in v1.1.1 

func (x *LLOStreamHistoryChunkProto) GetRecords() []*LLOStreamHistoryRecord

func (*LLOStreamHistoryChunkProto) GetSequence added in v1.1.1 

func (x *LLOStreamHistoryChunkProto) GetSequence() uint64

func (*LLOStreamHistoryChunkProto) ProtoMessage added in v1.1.1 

func (*LLOStreamHistoryChunkProto) ProtoMessage()

func (*LLOStreamHistoryChunkProto) ProtoReflect added in v1.1.1 

func (x *LLOStreamHistoryChunkProto) ProtoReflect() protoreflect.Message

func (*LLOStreamHistoryChunkProto) Reset added in v1.1.1 

func (x *LLOStreamHistoryChunkProto) Reset()

func (*LLOStreamHistoryChunkProto) String added in v1.1.1 

func (x *LLOStreamHistoryChunkProto) String() string

type LLOStreamHistoryHeaderProto added in v1.1.1 

type LLOStreamHistoryHeaderProto struct {

	// Capacity: the maximum depth required by any live channel/expression
	// referencing this pair. Zero means the pair is torn down.
	RequiredCount uint32 `protobuf:"varint,1,opt,name=requiredCount,proto3" json:"requiredCount,omitempty"`
	// Absolute sequence of the oldest retained chunk. Zero when nothing is
	// stored.
	FirstSequence uint64 `protobuf:"varint,2,opt,name=firstSequence,proto3" json:"firstSequence,omitempty"`
	// Records held by each retained chunk, oldest first. Every entry except the
	// last equals MaxHistoryChunkRecords; the last is in [1, MaxHistoryChunkRecords].
	Counts []uint32 `protobuf:"varint,3,rep,packed,name=counts,proto3" json:"counts,omitempty"`
	// observedAtNanoseconds of the first record of each retained chunk, parallel
	// to counts and strictly increasing. Denormalized so that evicting a chunk
	// does not require loading its successor to learn the window's new start.
	ChunkFirstObservationTimestampNanoseconds []uint64 `` /* 143-byte string literal not displayed */
	// observedAtNanoseconds of the newest record in the window, or 0 when empty.
	// The strictly-newer append rule is decided against this alone.
	LastObservationTimestampNanoseconds uint64 `protobuf:"varint,5,opt,name=lastObservationTimestampNanoseconds,proto3" json:"lastObservationTimestampNanoseconds,omitempty"`
	// contains filtered or unexported fields
}

LLOStreamHistoryHeaderProto is the index of a chunked history window: which chunks are retained, how full each one is, and when each one starts.

It is sized so that every decision a round makes — is there enough depth, which chunks must be read, may this value be appended, which chunk falls out — can be taken from the header alone, without reading a single chunk.

func (*LLOStreamHistoryHeaderProto) Descriptor deprecated added in v1.1.1 

func (*LLOStreamHistoryHeaderProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamHistoryHeaderProto.ProtoReflect.Descriptor instead.

func (*LLOStreamHistoryHeaderProto) GetChunkFirstObservationTimestampNanoseconds added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) GetChunkFirstObservationTimestampNanoseconds() []uint64

func (*LLOStreamHistoryHeaderProto) GetCounts added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) GetCounts() []uint32

func (*LLOStreamHistoryHeaderProto) GetFirstSequence added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) GetFirstSequence() uint64

func (*LLOStreamHistoryHeaderProto) GetLastObservationTimestampNanoseconds added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) GetLastObservationTimestampNanoseconds() uint64

func (*LLOStreamHistoryHeaderProto) GetRequiredCount added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) GetRequiredCount() uint32

func (*LLOStreamHistoryHeaderProto) ProtoMessage added in v1.1.1 

func (*LLOStreamHistoryHeaderProto) ProtoMessage()

func (*LLOStreamHistoryHeaderProto) ProtoReflect added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) ProtoReflect() protoreflect.Message

func (*LLOStreamHistoryHeaderProto) Reset added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) Reset()

func (*LLOStreamHistoryHeaderProto) String added in v1.1.1 

func (x *LLOStreamHistoryHeaderProto) String() string

type LLOStreamHistoryRecord added in v1.1.1 

type LLOStreamHistoryRecord struct {
	ObservedAtNanoseconds uint64          `protobuf:"varint,1,opt,name=observedAtNanoseconds,proto3" json:"observedAtNanoseconds,omitempty"`
	Value                 *LLOStreamValue `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
	// contains filtered or unexported fields
}

LLOStreamHistoryRecord is one agreed aggregate value of a stream, with the timestamp it was observed at. Per-record timestamps are required: rounds are not evenly spaced, so time-weighted functions and gap detection cannot work from the window bounds alone.

func (*LLOStreamHistoryRecord) Descriptor deprecated added in v1.1.1 

func (*LLOStreamHistoryRecord) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamHistoryRecord.ProtoReflect.Descriptor instead.

func (*LLOStreamHistoryRecord) GetObservedAtNanoseconds added in v1.1.1 

func (x *LLOStreamHistoryRecord) GetObservedAtNanoseconds() uint64

func (*LLOStreamHistoryRecord) GetValue added in v1.1.1 

func (x *LLOStreamHistoryRecord) GetValue() *LLOStreamValue

func (*LLOStreamHistoryRecord) ProtoMessage added in v1.1.1 

func (*LLOStreamHistoryRecord) ProtoMessage()

func (*LLOStreamHistoryRecord) ProtoReflect added in v1.1.1 

func (x *LLOStreamHistoryRecord) ProtoReflect() protoreflect.Message

func (*LLOStreamHistoryRecord) Reset added in v1.1.1 

func (x *LLOStreamHistoryRecord) Reset()

func (*LLOStreamHistoryRecord) String added in v1.1.1 

func (x *LLOStreamHistoryRecord) String() string

type LLOStreamObservationProto 

type LLOStreamObservationProto struct {
	Valid bool   `protobuf:"varint,1,opt,name=valid,proto3" json:"valid,omitempty"`
	Value []byte `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOStreamObservationProto) Descriptor deprecated 

func (*LLOStreamObservationProto) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamObservationProto.ProtoReflect.Descriptor instead.

func (*LLOStreamObservationProto) GetValid 

func (x *LLOStreamObservationProto) GetValid() bool

func (*LLOStreamObservationProto) GetValue 

func (x *LLOStreamObservationProto) GetValue() []byte

func (*LLOStreamObservationProto) ProtoMessage 

func (*LLOStreamObservationProto) ProtoMessage()

func (*LLOStreamObservationProto) ProtoReflect 

func (x *LLOStreamObservationProto) ProtoReflect() protoreflect.Message

func (*LLOStreamObservationProto) Reset 

func (x *LLOStreamObservationProto) Reset()

func (*LLOStreamObservationProto) String 

func (x *LLOStreamObservationProto) String() string

type LLOStreamValue 

type LLOStreamValue struct {
	Type  LLOStreamValue_Type `protobuf:"varint,1,opt,name=type,proto3,enum=v1.LLOStreamValue_Type" json:"type,omitempty"`
	Value []byte              `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
	// contains filtered or unexported fields
}

func StreamValueToProto 

func StreamValueToProto(v StreamValue) (*LLOStreamValue, error)

StreamValueToProto encodes a StreamValue into its protobuf wire form.

func (*LLOStreamValue) Descriptor deprecated 

func (*LLOStreamValue) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamValue.ProtoReflect.Descriptor instead.

func (*LLOStreamValue) GetType 

func (x *LLOStreamValue) GetType() LLOStreamValue_Type

func (*LLOStreamValue) GetValue 

func (x *LLOStreamValue) GetValue() []byte

func (*LLOStreamValue) ProtoMessage 

func (*LLOStreamValue) ProtoMessage()

func (*LLOStreamValue) ProtoReflect 

func (x *LLOStreamValue) ProtoReflect() protoreflect.Message

func (*LLOStreamValue) Reset 

func (x *LLOStreamValue) Reset()

func (*LLOStreamValue) String 

func (x *LLOStreamValue) String() string

type LLOStreamValueQuote 

type LLOStreamValueQuote struct {
	Bid       []byte `protobuf:"bytes,1,opt,name=bid,proto3" json:"bid,omitempty"`
	Benchmark []byte `protobuf:"bytes,2,opt,name=benchmark,proto3" json:"benchmark,omitempty"`
	Ask       []byte `protobuf:"bytes,3,opt,name=ask,proto3" json:"ask,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOStreamValueQuote) Descriptor deprecated 

func (*LLOStreamValueQuote) Descriptor() ([]byte, []int)

Deprecated: Use LLOStreamValueQuote.ProtoReflect.Descriptor instead.

func (*LLOStreamValueQuote) GetAsk 

func (x *LLOStreamValueQuote) GetAsk() []byte

func (*LLOStreamValueQuote) GetBenchmark 

func (x *LLOStreamValueQuote) GetBenchmark() []byte

func (*LLOStreamValueQuote) GetBid 

func (x *LLOStreamValueQuote) GetBid() []byte

func (*LLOStreamValueQuote) ProtoMessage 

func (*LLOStreamValueQuote) ProtoMessage()

func (*LLOStreamValueQuote) ProtoReflect 

func (x *LLOStreamValueQuote) ProtoReflect() protoreflect.Message

func (*LLOStreamValueQuote) Reset 

func (x *LLOStreamValueQuote) Reset()

func (*LLOStreamValueQuote) String 

func (x *LLOStreamValueQuote) String() string

type LLOStreamValue_Type 

type LLOStreamValue_Type int32
const (
	LLOStreamValue_Decimal                LLOStreamValue_Type = 0
	LLOStreamValue_Quote                  LLOStreamValue_Type = 1
	LLOStreamValue_TimestampedStreamValue LLOStreamValue_Type = 2
)

func (LLOStreamValue_Type) Descriptor 

func (LLOStreamValue_Type) Descriptor() protoreflect.EnumDescriptor

func (LLOStreamValue_Type) Enum 

func (x LLOStreamValue_Type) Enum() *LLOStreamValue_Type

func (LLOStreamValue_Type) EnumDescriptor deprecated 

func (LLOStreamValue_Type) EnumDescriptor() ([]byte, []int)

Deprecated: Use LLOStreamValue_Type.Descriptor instead.

func (LLOStreamValue_Type) Number 

func (x LLOStreamValue_Type) Number() protoreflect.EnumNumber

func (LLOStreamValue_Type) String 

func (x LLOStreamValue_Type) String() string

func (LLOStreamValue_Type) Type 

func (LLOStreamValue_Type) Type() protoreflect.EnumType

type LLOTimestampedStreamValue 

type LLOTimestampedStreamValue struct {
	ObservedAtNanoseconds uint64          `protobuf:"varint,1,opt,name=observedAtNanoseconds,proto3" json:"observedAtNanoseconds,omitempty"`
	StreamValue           *LLOStreamValue `protobuf:"bytes,2,opt,name=streamValue,proto3" json:"streamValue,omitempty"`
	// contains filtered or unexported fields
}

func (*LLOTimestampedStreamValue) Descriptor deprecated 

func (*LLOTimestampedStreamValue) Descriptor() ([]byte, []int)

Deprecated: Use LLOTimestampedStreamValue.ProtoReflect.Descriptor instead.

func (*LLOTimestampedStreamValue) GetObservedAtNanoseconds 

func (x *LLOTimestampedStreamValue) GetObservedAtNanoseconds() uint64

func (*LLOTimestampedStreamValue) GetStreamValue 

func (x *LLOTimestampedStreamValue) GetStreamValue() *LLOStreamValue

func (*LLOTimestampedStreamValue) ProtoMessage 

func (*LLOTimestampedStreamValue) ProtoMessage()

func (*LLOTimestampedStreamValue) ProtoReflect 

func (x *LLOTimestampedStreamValue) ProtoReflect() protoreflect.Message

func (*LLOTimestampedStreamValue) Reset 

func (x *LLOTimestampedStreamValue) Reset()

func (*LLOTimestampedStreamValue) String 

func (x *LLOTimestampedStreamValue) String() string

type OffchainConfig 

type OffchainConfig struct {
	ProtocolVersion uint32
	// DefaultMinReportIntervalNanoseconds is the default minimum report interval in nanoseconds.
	// It must be set to 0 for protocol version 0.
	// It must be set to 1 or greater for protocol version 1+.
	//
	// NOTE: This merely controls the _minimum_ interval between reports. It
	// does not guarantee a maximum interval. If you want reports to be
	// produced quickly, you are still limited by OCR3's DeltaRound and
	// DeltaGrace params, as well as networking latency.
	DefaultMinReportIntervalNanoseconds uint64
	// EnableObservationCompression enables observation compression.
	EnableObservationCompression bool
}

func DecodeOffchainConfig 

func DecodeOffchainConfig(b []byte) (o OffchainConfig, err error)

func (OffchainConfig) Encode 

func (c OffchainConfig) Encode() ([]byte, error)

func (OffchainConfig) Validate 

func (c OffchainConfig) Validate() error

type OnchainConfig 

type OnchainConfig struct {
	Version                 uint8
	PredecessorConfigDigest *types.ConfigDigest
}

type OnchainConfigCodec 

type OnchainConfigCodec interface {
	Decode(b []byte) (OnchainConfig, error)
	Encode(OnchainConfig) ([]byte, error)
}

type OptsCache 

type OptsCache struct {
	// contains filtered or unexported fields
}

OptsCache caches decoded channel definition options keyed by (ChannelID, target type). Raw opts bytes are stored via Set during channel definition changes in Outcome(). Decoded values are produced lazily on the first GetOpts call for a given (channelID, type) and reused until the channel is updated or removed.

func NewOptsCache 

func NewOptsCache() *OptsCache

func (*OptsCache) Len 

func (c *OptsCache) Len() int

Len returns the number of channels in the cache.

func (*OptsCache) Remove 

func (c *OptsCache) Remove(channelID llotypes.ChannelID)

Remove removes all raw and decoded data for a channel.

func (*OptsCache) ResetTo 

func (c *OptsCache) ResetTo(channelDefinitions llotypes.ChannelDefinitions)

ResetTo resets the cache to the given channel definitions.

func (*OptsCache) Set 

func (c *OptsCache) Set(channelID llotypes.ChannelID, raw llotypes.ChannelOpts)

Set stores the raw opts for a channel and invalidates any previously decoded values for that channel. It is a no-op when the raw bytes are identical to what is already stored.

type PredecessorRetirementReportCache 

type PredecessorRetirementReportCache interface {
	// AttestedRetirementReport returns the attested retirement report for the
	// given config digest from the local cache.
	//
	// This should return nil and not error in the case of a missing attested
	// retirement report.
	AttestedRetirementReport(predecessorConfigDigest ocr2types.ConfigDigest) ([]byte, error)
	// CheckAttestedRetirementReport verifies that an attested retirement
	// report, which may have come from another node, is valid (signed) with
	// signers corresponding to the given config digest
	CheckAttestedRetirementReport(predecessorConfigDigest ocr2types.ConfigDigest, attestedRetirementReport []byte) (RetirementReport, error)
}

The predecessor protocol instance stores its attested retirement report in this cache (locally, offchain), so it can be fetched by the successor protocol instance.

PredecessorRetirementReportCache is populated by the old protocol instance writing to it and the new protocol instance reading from it.

The sketch envisions it being implemented as a single object that is shared between different protocol instances.

type Quote 

type Quote struct {
	Bid       decimal.Decimal
	Benchmark decimal.Decimal
	Ask       decimal.Decimal
}

func (*Quote) IsValid 

func (v *Quote) IsValid() bool

func (*Quote) MarshalBinary 

func (v *Quote) MarshalBinary() (b []byte, err error)

func (*Quote) MarshalText 

func (v *Quote) MarshalText() ([]byte, error)

func (*Quote) Type 

func (v *Quote) Type() LLOStreamValue_Type

func (*Quote) UnmarshalBinary 

func (v *Quote) UnmarshalBinary(data []byte) error

func (*Quote) UnmarshalText 

func (v *Quote) UnmarshalText(data []byte) error

type Report 

type Report struct {
	ConfigDigest types.ConfigDigest
	// OCR sequence number of this report
	SeqNr uint64
	// Channel that is being reported on
	ChannelID llotypes.ChannelID
	// Report is only valid at t > ValidAfterNanoseconds
	// ValidAfterNanoseconds < ObservationTimestampNanoseconds always, by enforcement
	// in IsReportable
	ValidAfterNanoseconds uint64
	// ObservationTimestampNanoseconds is the median of all observation timestamps
	// (note that this timestamp is taken immediately before we initiate any
	// observations)
	ObservationTimestampNanoseconds uint64
	// Values for every stream in the channel
	Values []StreamValue
	// The contract onchain will only validate non-specimen reports. A staging
	// protocol instance will generate specimen reports so we can validate it
	// works properly without any risk of misreports landing on chain.
	Specimen bool
}

type ReportCodec 

type ReportCodec interface {
	// Encode may be lossy, so no Decode function is expected
	// Encode should handle nil stream aggregate values without panicking (it
	// may return error instead).
	// Codecs may use GetOpts(optsCache, report.ChannelID) to get cached parsed opts.
	Encode(Report, llotypes.ChannelDefinition, *OptsCache) ([]byte, error)
	// Verify may optionally verify a channel definition to ensure it is valid
	// for the given report codec. If a codec does not wish to implement
	// validation it may simply return nil here. If any definition fails
	// validation, the entire channel definitions file will be rejected.
	// This can be useful to ensure that e.g. options aren't changed
	// accidentally to something that would later break a report on encoding.
	Verify(llotypes.ChannelDefinition) error
}

type ReportCodecHistoryBackfill 

type ReportCodecHistoryBackfill struct{}

ReportCodecHistoryBackfill validates channel definitions; encoding is delegated to the target channel codec.

func (ReportCodecHistoryBackfill) Encode 

func (ReportCodecHistoryBackfill) Encode(Report, llotypes.ChannelDefinition, *OptsCache) ([]byte, error)

func (ReportCodecHistoryBackfill) Verify 

func (ReportCodecHistoryBackfill) Verify(cd llotypes.ChannelDefinition) error

type RetirementReport 

type RetirementReport struct {
	// Retirement reports are not guaranteed to be compatible across different
	// protocol versions
	ProtocolVersion uint32
	// Carries validity time stamps between protocol instances to ensure there
	// are no gaps
	ValidAfterNanoseconds map[llotypes.ChannelID]uint64
}

type RetirementReportCodec 

type RetirementReportCodec interface {
	Encode(RetirementReport) ([]byte, error)
	Decode([]byte) (RetirementReport, error)
}

type RingWindow added in v1.1.1 

type RingWindow struct {
	// contains filtered or unexported fields
}

RingWindow is the per-round working copy of one pair's chunked history.

Chunks are supplied by the caller rather than read by this type, which is what keeps the layout logic free of KV types and independently testable. The protocol is: 

w := NewRingWindow(header)          // header read from storage, nil if none
for _, seq := range w.AppendPlan()  // and/or w.ReadPlan(n)
    w.Provide(chunk)                // chunks read from storage
w.Append(...) / w.Newest(n)         // mutate and read
w.WriteSet()                        // what to persist

AppendPlan and ReadPlan name chunks by absolute sequence; HistoryChunkSlot turns those into the keys to read.

func NewRingWindow added in v1.1.1 

func NewRingWindow(header *StreamHistoryHeader) *RingWindow

NewRingWindow returns a working copy over an existing header. A nil header means nothing is stored yet.

func ResetRingWindow added in v1.1.1 

func ResetRingWindow() *RingWindow

ResetRingWindow returns an empty working copy whose write set deletes every slot of the ring.

This is the recovery path for a window that failed to decode. The header is the only thing that says which chunks exist, so when it is unusable the only safe move is to delete the whole slot space — which is bounded, and is the reason the layout uses a fixed ring rather than an unbounded sequence. The cost of recovery is a warmup, never a halted round.

func (*RingWindow) Append added in v1.1.1 

func (w *RingWindow) Append(observedAtNanoseconds uint64, value StreamValue) (appended bool, err error)

Append adds a value to the newest end of the window, sealing the newest chunk and starting another when it fills, and evicting whole chunks from the oldest end once they are no longer needed. It reports whether the record was stored.

A record is stored only if its timestamp is strictly newer than the current newest. That single rule keeps the series monotonic and prevents the two ways a value could otherwise be double counted: a carry-forward timestamped aggregate re-appended every round until it refreshes, and a non-advancing or regressing consensus observation timestamp. Neither is an error — both are normal — so a rejected append returns (false, nil).

A pair with zero capacity stores nothing. A value serializing to more than MaxHistoryRecordBytes is rejected with ErrHistoryRecordTooLarge, leaving an honest gap rather than a window larger than the byte budget assumed.

At most one record may be stored per round, and a second attempt returns ErrHistoryAlreadyAppended. Two rules depend on it. The write set carries the newest chunk only, so a second append that sealed a chunk and started another would leave the sealed one's final record unpersisted while the header already counted it -- next round the window reads short and is discarded as corrupt. And the round's write budget is sized on one header and one chunk per pair. The aggregation path appends once per pair per round, so this only rejects misuse, but it is checked rather than assumed because Append is exported.

func (*RingWindow) AppendPlan added in v1.1.1 

func (w *RingWindow) AppendPlan() []uint64

AppendPlan returns the sequences that must be provided before Append can succeed: the newest chunk, when there is one with room left. Appending into a sealed or absent chunk starts a new one and needs nothing loaded.

func (*RingWindow) FirstObservationTimestampNanoseconds added in v1.1.1 

func (w *RingWindow) FirstObservationTimestampNanoseconds() uint64

FirstObservationTimestampNanoseconds is the timestamp of the oldest retained record, or zero when empty.

func (*RingWindow) Header added in v1.1.1 

func (w *RingWindow) Header() *StreamHistoryHeader

Header returns the window's current header.

func (*RingWindow) LastObservationTimestampNanoseconds added in v1.1.1 

func (w *RingWindow) LastObservationTimestampNanoseconds() uint64

LastObservationTimestampNanoseconds is the timestamp of the newest record, or zero when empty.

func (*RingWindow) Len added in v1.1.1 

func (w *RingWindow) Len() int

Len is the number of records retained, which may overshoot RequiredCount by less than one chunk.

func (*RingWindow) Loaded added in v1.1.1 

func (w *RingWindow) Loaded(sequence uint64) bool

Loaded reports whether a chunk has already been provided.

Callers must consult this before re-reading a planned sequence: a chunk the window has already been given may have been appended to since, so the stored bytes are stale until the write set is flushed and providing them again would look like corruption.

func (*RingWindow) Newest added in v1.1.1 

func (w *RingWindow) Newest(n uint32) ([]StreamHistoryRecord, error)

Newest returns the n most recent records, oldest first, from the chunks ReadPlan named. It returns ErrInsufficientStreamHistory if fewer than n are retained, and ErrHistoryChunkNotLoaded if a planned chunk was not provided.

func (*RingWindow) Provide added in v1.1.1 

func (w *RingWindow) Provide(chunk *StreamHistoryChunk) error

Provide hands a decoded chunk to the window, checking it against the header.

This is where a stale chunk is caught: reads are by ring slot, so a slot left behind by an earlier lap decodes fine but carries a sequence the header no longer retains. Treating that as corruption — rather than as data — is what makes slot reuse safe.

func (*RingWindow) ReadPlan added in v1.1.1 

func (w *RingWindow) ReadPlan(n uint32) ([]uint64, error)

ReadPlan returns the sequences covering the newest n records, oldest first.

It returns ErrInsufficientStreamHistory when fewer than n records are retained — decided from the header, so a warming-up pair costs no chunk reads at all. A short window is never silently substituted.

func (*RingWindow) RequiredCount added in v1.1.1 

func (w *RingWindow) RequiredCount() uint32

RequiredCount is the window capacity.

func (*RingWindow) SetRequiredCount added in v1.1.1 

func (w *RingWindow) SetRequiredCount(requiredCount uint32) (changed bool, err error)

SetRequiredCount updates the capacity, evicting chunks the new depth no longer needs. It reports whether anything changed, so callers can avoid writing an unmodified header.

Growing the capacity does not synthesize records and touches no chunk: the extra depth fills over subsequent rounds, and expressions needing it stay unsatisfied meanwhile. Shrinking is deletes only — a sealed chunk is never rewritten. Setting zero tears the window down, evicting everything.

func (*RingWindow) WriteSet added in v1.1.1 

func (w *RingWindow) WriteSet() RingWriteSet

WriteSet returns what this round's mutations mean for storage: at most one header, at most one chunk — the newest, the only one a round ever rewrites — and the slots of any chunks evicted.

type RingWriteSet added in v1.1.1 

type RingWriteSet struct {
	// Header is the header to write, or nil when it did not change.
	Header *StreamHistoryHeader
	// Chunk is the newest chunk, the only one a round ever rewrites, or nil
	// when nothing was appended.
	Chunk *StreamHistoryChunk
	// DeletedSlots are ring slots to delete, ascending: chunks evicted this
	// round, or every slot when the window was reset.
	DeletedSlots []uint32
}

RingWriteSet is what one round's mutations mean for storage. It is produced by RingWindow.WriteSet and executed by the plugin, which owns the KV types.

Writes must be applied before deletes. A slot is never both written and deleted in the same round — WriteSet drops the tail slot from DeletedSlots if it somehow appears — but ordering the two makes that independent of the caller getting the argument order right.

func (RingWriteSet) Empty added in v1.1.1 

func (s RingWriteSet) Empty() bool

Empty reports whether the round changed nothing about this window.

type StandardRetirementReportCodec 

type StandardRetirementReportCodec struct{}

func (StandardRetirementReportCodec) Decode 

func (r StandardRetirementReportCodec) Decode(data []byte) (RetirementReport, error)

func (StandardRetirementReportCodec) Encode 

func (r StandardRetirementReportCodec) Encode(report RetirementReport) ([]byte, error)

type StreamAggregates 

type StreamAggregates map[llotypes.StreamID]map[llotypes.Aggregator]StreamValue

type StreamHistoryChunk added in v1.1.1 

type StreamHistoryChunk struct {
	// contains filtered or unexported fields
}

StreamHistoryChunk is one slot of a chunked history ring: a run of consecutive records for one (streamID, aggregator) pair, oldest first.

Only the newest chunk of a window is ever rewritten. Once a chunk holds MaxHistoryChunkRecords records it is sealed: immutable for as long as it is retained, and never part of a write set again. That is what makes per-round write cost a function of the chunk size rather than of the window depth.

The records slice is unexported so a caller cannot assemble a chunk that violates the invariants the decoder enforces.

func StreamHistoryChunkFromProto added in v1.1.1 

func StreamHistoryChunkFromProto(pb *LLOStreamHistoryChunkProto) (*StreamHistoryChunk, error)

StreamHistoryChunkFromProto validates and converts a decoded chunk.

Everything checkable without the header is checked here; the header-relative checks (is this sequence still retained, does it hold the number of records the header claims) happen in RingWindow.Provide.

func UnmarshalStreamHistoryChunk added in v1.1.1 

func UnmarshalStreamHistoryChunk(data []byte) (*StreamHistoryChunk, error)

UnmarshalStreamHistoryChunk decodes and validates a stored chunk.

func (*StreamHistoryChunk) Len added in v1.1.1 

func (c *StreamHistoryChunk) Len() int

Len is the number of records in the chunk.

func (*StreamHistoryChunk) MarshalBinary added in v1.1.1 

func (c *StreamHistoryChunk) MarshalBinary() ([]byte, error)

MarshalBinary serializes the chunk deterministically for storage.

func (*StreamHistoryChunk) Records added in v1.1.1 

func (c *StreamHistoryChunk) Records() []StreamHistoryRecord

Records returns the chunk's records, oldest first. The slice aliases internal state and must be treated as read-only; it is not copied because this is on the per-round hot path.

func (*StreamHistoryChunk) Sequence added in v1.1.1 

func (c *StreamHistoryChunk) Sequence() uint64

Sequence is the chunk's absolute index in the window, monotonically increasing over the window's life. The key holds only the slot (sequence mod MaxHistoryChunkSlots), so this is what tells a live chunk apart from a stale one left behind by an earlier lap of the ring.

func (*StreamHistoryChunk) Slot added in v1.1.1 

func (c *StreamHistoryChunk) Slot() uint32

Slot is the ring slot this chunk occupies.

func (*StreamHistoryChunk) ToProto added in v1.1.1 

func (c *StreamHistoryChunk) ToProto() (*LLOStreamHistoryChunkProto, error)

ToProto converts the chunk to its wire form.

type StreamHistoryHeader added in v1.1.1 

type StreamHistoryHeader struct {
	// contains filtered or unexported fields
}

StreamHistoryHeader is the index of a chunked history window: which chunks are retained, how full each one is, and when each one starts.

It is deliberately sufficient on its own. Every decision a round makes — is there enough depth, which chunks must be read, may this value be appended, which chunk falls out — is taken from the header without reading a chunk, so a pair that is still warming up costs exactly one read for the whole round.

Invariants, enforced on decode and by every mutator:

  • len(counts) == len(chunkFirst) <= MaxHistoryChunkSlots.
  • counts[i] == MaxHistoryChunkRecords for every i but the last, which is in [1, MaxHistoryChunkRecords].
  • chunkFirst is strictly increasing, and chunkFirst[last] <= last.
  • total() < requiredCount + MaxHistoryChunkRecords, and dropping the oldest chunk would leave fewer than requiredCount records — retention keeps as few whole chunks as cover the required depth.
  • an empty window has firstSequence == 0 and last == 0.

func StreamHistoryHeaderFromProto added in v1.1.1 

func StreamHistoryHeaderFromProto(pb *LLOStreamHistoryHeaderProto) (*StreamHistoryHeader, error)

StreamHistoryHeaderFromProto validates and converts a decoded header.

Every check guards against corrupt or byzantine stored state, and every failure is ErrCorruptStreamHistory so callers can uniformly discard the whole window and re-warm. A header that decodes is trustworthy enough to plan reads and evictions from without cross-checking it against the chunks.

func UnmarshalStreamHistoryHeader added in v1.1.1 

func UnmarshalStreamHistoryHeader(data []byte) (*StreamHistoryHeader, error)

UnmarshalStreamHistoryHeader decodes and validates a stored header.

func (*StreamHistoryHeader) ChunkCount added in v1.1.1 

func (h *StreamHistoryHeader) ChunkCount() int

ChunkCount is the number of retained chunks.

func (*StreamHistoryHeader) Counts added in v1.1.1 

func (h *StreamHistoryHeader) Counts() []uint32

Counts returns a copy of the per-chunk record counts, oldest first.

func (*StreamHistoryHeader) FirstObservationTimestampNanoseconds added in v1.1.1 

func (h *StreamHistoryHeader) FirstObservationTimestampNanoseconds() uint64

FirstObservationTimestampNanoseconds is the timestamp of the oldest retained record, or zero when the window is empty.

func (*StreamHistoryHeader) FirstSequence added in v1.1.1 

func (h *StreamHistoryHeader) FirstSequence() uint64

FirstSequence is the absolute sequence of the oldest retained chunk.

func (*StreamHistoryHeader) LastObservationTimestampNanoseconds added in v1.1.1 

func (h *StreamHistoryHeader) LastObservationTimestampNanoseconds() uint64

LastObservationTimestampNanoseconds is the timestamp of the newest record, or zero when the window is empty. The strictly-newer append rule is decided against this alone, so appending costs no chunk read beyond the tail.

func (*StreamHistoryHeader) Len added in v1.1.1 

func (h *StreamHistoryHeader) Len() int

Len is the number of records the window holds across all retained chunks. It may exceed RequiredCount by up to one chunk, because retention works in whole chunks; readers ask for an exact depth and never see the overshoot.

func (*StreamHistoryHeader) MarshalBinary added in v1.1.1 

func (h *StreamHistoryHeader) MarshalBinary() ([]byte, error)

MarshalBinary serializes the header deterministically for storage.

func (*StreamHistoryHeader) RequiredCount added in v1.1.1 

func (h *StreamHistoryHeader) RequiredCount() uint32

RequiredCount is the window capacity: the deepest history any live channel requires for this pair. Zero means the pair is torn down.

func (*StreamHistoryHeader) Sequences added in v1.1.1 

func (h *StreamHistoryHeader) Sequences() []uint64

Sequences returns the absolute sequences of the retained chunks, oldest first.

func (*StreamHistoryHeader) ToProto added in v1.1.1 

func (h *StreamHistoryHeader) ToProto() *LLOStreamHistoryHeaderProto

ToProto converts the header to its wire form.

type StreamHistoryRecord added in v1.1.1 

type StreamHistoryRecord struct {
	ObservedAtNanoseconds uint64
	Value                 StreamValue
}

StreamHistoryRecord is one agreed aggregate value of a stream together with the timestamp it was observed at.

type StreamValue 

type StreamValue interface {
	// Binary marshaler/unmarshaler used for protobufs
	// Unmarshal should NOT panic on nil receiver, but instead return ErrNilStreamValue
	encoding.BinaryMarshaler
	encoding.BinaryUnmarshaler
	// TextMarshaler needed for JSON serialization and logging
	// Unmarshal should NOT panic on nil receiver, but instead return ErrNilStreamValue
	encoding.TextMarshaler
	encoding.TextUnmarshaler
	// Type is needed for proto serialization so we know how to unserialize it
	Type() LLOStreamValue_Type
}

func BuildBackfillStreamValues 

func BuildBackfillStreamValues(target llotypes.ChannelDefinition, row map[llotypes.StreamID]string) ([]StreamValue, error)

BuildBackfillStreamValues builds stream values for a backfill timestamp row in target stream order.

func MedianAggregator 

func MedianAggregator(values []StreamValue, f int) (StreamValue, error)

func ModeAggregator 

func ModeAggregator(values []StreamValue, f int) (StreamValue, error)

ModeAggregator works on arbitrary StreamValue types It picks the most common value There must be at least f+1 observations in agreement in order to produce a value nil observations are ignored

func QuoteAggregator 

func QuoteAggregator(values []StreamValue, f int) (StreamValue, error)

func StreamValueFromBackfillString 

func StreamValueFromBackfillString(agg llotypes.Aggregator, s string) (StreamValue, error)

StreamValueFromBackfillString parses a backfill observation string for the given aggregator.

func UnmarshalProtoStreamValue 

func UnmarshalProtoStreamValue(enc *LLOStreamValue) (sv StreamValue, err error)

func UnmarshalTypedTextStreamValue 

func UnmarshalTypedTextStreamValue(enc *TypedTextStreamValue) (StreamValue, error)

type StreamValues 

type StreamValues map[llotypes.StreamID]StreamValue

Values for a set of streams, e.g. "eth-usd", "link-usd", "eur-chf" etc StreamIDs are uint32

type TimeResolution 

type TimeResolution uint8

TimeResolution represents the resolution for timestamp conversion 

const (
	ResolutionSeconds TimeResolution = iota
	ResolutionMilliseconds
	ResolutionMicroseconds
	ResolutionNanoseconds
)

func TargetChannelTimeResolution 

func TargetChannelTimeResolution(target llotypes.ChannelDefinition) (TimeResolution, error)

func (TimeResolution) MarshalJSON 

func (tp TimeResolution) MarshalJSON() ([]byte, error)

func (*TimeResolution) UnmarshalJSON 

func (tp *TimeResolution) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals TimeResolution from JSON - used to unmarshal from the Opts structs.

type TimestampedStreamValue 

type TimestampedStreamValue struct {
	ObservedAtNanoseconds uint64      `json:"observedAtNanoseconds"`
	StreamValue           StreamValue `json:"streamValue"`
}

TimestampedStreamValue is a StreamValue with an associated timestamp

func (*TimestampedStreamValue) MarshalBinary 

func (v *TimestampedStreamValue) MarshalBinary() ([]byte, error)

func (*TimestampedStreamValue) MarshalText 

func (v *TimestampedStreamValue) MarshalText() ([]byte, error)

func (*TimestampedStreamValue) Type 

func (v *TimestampedStreamValue) Type() LLOStreamValue_Type

func (*TimestampedStreamValue) UnmarshalBinary 

func (v *TimestampedStreamValue) UnmarshalBinary(data []byte) error

func (*TimestampedStreamValue) UnmarshalText 

func (v *TimestampedStreamValue) UnmarshalText(data []byte) error

type TypedTextStreamValue 

type TypedTextStreamValue struct {
	Type                  LLOStreamValue_Type `json:"t"`
	SerializedStreamValue string              `json:"v"`
}

func NewTypedTextStreamValue 

func NewTypedTextStreamValue(sv StreamValue) (TypedTextStreamValue, error)

Source Files

View all Source files

Directories

Path Synopsis
Package calculated evaluates the expression language used by EVMABIEncodeUnpackedExpr channels to derive new stream values from observed ones.
 Package calculated evaluates the expression language used by EVMABIEncodeUnpackedExpr channels to derive new stream values from observed ones.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.