llo/dev/v31: limits, fixes and documentation

llo/datasource/datasource.go

@@ -60,6 +60,12 @@ func (o *dsOpts) LifeCycleStage() llotypes.LifeCycleStage { return o.lifeCycleSt
 // DataSource observes stream values. For each known streamID, Observe should
 // set the observed value in the passed protocol.StreamValues. Unknown/failed streams
 // should be left unset. Shared across all LLO plugin versions.
+//
+// Observe owns streamValues only until it returns, including when it returns
+// because ctx is done. The caller reads and serializes the map right after, so
+// an implementation that fills it from other goroutines must stop them, or stop
+// them writing, before returning. Observe should not add streams that are not
+// requested.
 type DataSource interface {
 	Observe(ctx context.Context, streamValues protocol.StreamValues, opts DSOpts) error
 }

llo/dev/v31/backfill.go

@@ -1,6 +1,8 @@
 package llo
 
 import (
+	"sort"
+
 	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
 
 	protocol "github.com/smartcontractkit/chainlink-data-streams/llo/protocol"
@@ -9,37 +11,48 @@ import (
 // selectBackfillCandidate returns the next history-backfill observation to emit
 // for backfillCID: its timestamp (nanoseconds), the raw opts key, and the parsed
 // opts. ok is false when the channel is not a valid, in-progress backfill (no
-// definition, wrong format, missing target, no watermark, or backfill complete).
+// definition, wrong format, missing or tombstoned target, no watermark, or
+// backfill complete).
 //
 // The watermark is validAfter[backfillCID] (the last emitted backfill observation
-// time); the next candidate is the smallest observation strictly after the
-// watermark and strictly before obsTsNanos. This mirrors the v30
+// time); the next candidate is the smallest emittable observation strictly after
+// the watermark and strictly before obsTsNanos. This mirrors the v30
 // SelectBackfillCandidate but operates on plain fields (shared by precursor and
 // kvState) and returns a bool instead of an UnreportableChannelError.
-func selectBackfillCandidate(defs llotypes.ChannelDefinitions, validAfter map[llotypes.ChannelID]uint64, obsTsNanos uint64, backfillCID llotypes.ChannelID, optsCache *protocol.OptsCache) (tsNanos uint64, rawTS uint64, opts protocol.HistoryBackfillOpts, ok bool) {
+//
+// skipped counts the earlier rows in the window passed over as unemittable.
+// Selecting the next row consumes them: the watermark advances past them.
+func selectBackfillCandidate(defs llotypes.ChannelDefinitions, validAfter map[llotypes.ChannelID]uint64, obsTsNanos uint64, backfillCID llotypes.ChannelID, optsCache *protocol.OptsCache) (tsNanos uint64, rawTS uint64, opts protocol.HistoryBackfillOpts, skipped int, ok bool) {
 	cd, exists := defs[backfillCID]
 	if !exists || cd.ReportFormat != llotypes.ReportFormatHistoryBackfill {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+		return 0, 0, protocol.HistoryBackfillOpts{}, 0, false
 	}
 	o, err := protocol.GetHistoryBackfillOpts(optsCache, cd, backfillCID)
 	if err != nil {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+		return 0, 0, protocol.HistoryBackfillOpts{}, 0, false
 	}
+	// A tombstoned target no longer reports, so its history is not backfilled
+	// either. The watermark holds, and backfill resumes if the target returns.
 	target, exists := defs[o.TargetChannelID]
-	if !exists {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+	if !exists || target.Tombstone {
+		return 0, 0, protocol.HistoryBackfillOpts{}, 0, false
 	}
 	res, err := protocol.TargetChannelTimeResolution(target)
 	if err != nil {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+		return 0, 0, protocol.HistoryBackfillOpts{}, 0, false
+	}
+	// Reports needs the target's report-timestamp resolution. It depends on the
+	// target alone, so when it fails no row can be emitted.
+	if _, err := protocol.ReportTimestampResolutionNanos(target); err != nil {
+		return 0, 0, protocol.HistoryBackfillOpts{}, 0, false
 	}
 	watermark, exists := validAfter[backfillCID]
 	if !exists {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+		return 0, 0, protocol.HistoryBackfillOpts{}, 0, false
 	}
 
-	var bestRaw, bestNanos uint64
-	found := false
+	type candidate struct{ raw, nanos uint64 }
+	var candidates []candidate
 	for rawKey := range o.Observations {
 		tsN, ok := protocol.ObservationTimestampKeyToNanoseconds(rawKey, res)
 		if !ok {
@@ -50,26 +63,32 @@ func selectBackfillCandidate(defs llotypes.ChannelDefinitions, validAfter map[ll
 		if tsN >= obsTsNanos || tsN <= watermark {
 			continue
 		}
-		if !found || tsN < bestNanos || (tsN == bestNanos && rawKey < bestRaw) {
-			found = true
-			bestNanos = tsN
-			bestRaw = rawKey
-		}
-	}
-	if !found {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+		candidates = append(candidates, candidate{raw: rawKey, nanos: tsN})
 	}
-	// The candidate must also be emittable, not merely selectable. Reports
-	// needs the target's report-timestamp resolution and the row's stream
-	// values; if either fails there, the report is skipped while the watermark
-	// has already advanced past the row in the state transition, losing it
-	// permanently. Both are pure functions of (target definition, row), so
-	// checking them here keeps selection and emission on one path.
-	if _, err := protocol.ReportTimestampResolutionNanos(target); err != nil {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
-	}
-	if _, err := protocol.BuildBackfillStreamValues(target, o.Observations[bestRaw]); err != nil {
-		return 0, 0, protocol.HistoryBackfillOpts{}, false
+	sort.Slice(candidates, func(i, j int) bool {
+		if candidates[i].nanos != candidates[j].nanos {
+			return candidates[i].nanos < candidates[j].nanos
+		}
+		return candidates[i].raw < candidates[j].raw
+	})
+	// The candidate must be emittable, not merely selectable, or Reports skips
+	// it after the watermark has already advanced past it. Building the row's
+	// stream values is a pure function of (target definition, row), so checking
+	// it here keeps selection and emission on one path. A row that cannot be
+	// built never will be: pass over it rather than stall the backfill on it.
+	for _, c := range candidates {
+		values, err := protocol.BuildBackfillStreamValues(target, o.Observations[c.raw])
+		if err != nil {
+			skipped++
+			continue
+		}
+		// Backfill rows are fixed in the opts, so a value outside its encoded
+		// range never comes back into it.
+		if protocol.CheckEncodableValues(optsCache, target, o.TargetChannelID, values) != nil {
+			skipped++
+			continue
+		}
+		return c.nanos, c.raw, o, skipped, true
 	}
-	return bestNanos, bestRaw, o, true
+	return 0, 0, protocol.HistoryBackfillOpts{}, skipped, false
 }

llo/dev/v31/blobcompress.go

@@ -57,9 +57,12 @@ func getBlobCompressor() (blobCompressor, error) {
 			return
 		}
 		// WithDecoderMaxMemory bounds the decompressed size to a safe limit.
+		// WithDecodeAllCapLimit bounds each DecodeAll call to the capacity of
+		// its destination, which decodeBlobPayload sizes to the call's budget.
 		decoder, err := zstd.NewReader(nil,
 			zstd.WithDecoderConcurrency(0),
 			zstd.WithDecoderMaxMemory(maxDecompressedBlobPayloadBytes),
+			zstd.WithDecodeAllCapLimit(true),
 		)
 		if err != nil {
 			blobZstdErr = fmt.Errorf("new zstd decoder: %w", err)
@@ -101,9 +104,10 @@ func encodeBlobPayload(raw []byte) ([]byte, error) {
 
 // decodeBlobPayload reverses encodeBlobPayload. The payload is untrusted, so
 // the decompressed size is bounded by maxOut: cheaply up front via the zstd
-// frame header when it declares a content size, and again after the fact in
-// case it does not. maxOut is a caller-supplied budget (see decodeObservation)
-// and is clamped to maxDecompressedBlobPayloadBytes.
+// frame header when it declares a content size, and during decompression via
+// the capacity of the destination buffer in case it does not. maxOut is a
+// caller-supplied budget (see decodeObservation) and is clamped to
+// maxDecompressedBlobPayloadBytes.
 func decodeBlobPayload(payload []byte, maxOut int) ([]byte, error) {
 	if len(payload) == 0 {
 		return nil, fmt.Errorf("empty blob payload")
@@ -136,12 +140,19 @@ func decodeBlobPayload(payload []byte, maxOut int) ([]byte, error) {
 		if hdr.HasFCS && hdr.FrameContentSize > uint64(maxOut) {
 			return nil, fmt.Errorf("declared decompressed blob payload too large: %d > %d bytes", hdr.FrameContentSize, maxOut)
 		}
-		out, err := c.decoder.DecodeAll(body, nil)
+		// The decoder stops once the output outgrows the destination capacity,
+		// so the per-call budget holds while decompressing, not only after.
+		// encodeBlobPayload always declares the content size, so an honest
+		// payload allocates exactly its size here.
+		capacity := maxOut
+		if hdr.HasFCS {
+			capacity = int(hdr.FrameContentSize) //nolint:gosec // bounded by maxOut above
+		}
+		out, err := c.decoder.DecodeAll(body, make([]byte, 0, capacity))
 		if err != nil {
 			return nil, fmt.Errorf("decompress blob payload: %w", err)
 		}
-		// The shared decoder only enforces the process-wide ceiling, so this
-		// check is what enforces the (possibly tighter) per-call budget.
+		// The cap limit is checked per block, so a block can overshoot it.
 		if len(out) > maxOut {
 			return nil, fmt.Errorf("decompressed blob payload too large: %d > %d bytes", len(out), maxOut)
 		}

llo/dev/v31/blobcompress_test.go

@@ -1,6 +1,7 @@
 package llo
 
 import (
+	"bytes"
 	"strings"
 	"testing"
 
@@ -108,6 +109,40 @@ func Test_decodeBlobPayload_Errors(t *testing.T) {
 	})
 }
 
+func Test_decodeBlobPayload_UndeclaredSize(t *testing.T) {
+	// A streaming encoder does not know the content size up front, so the frame
+	// omits it and the header check cannot reject the payload.
+	var buf bytes.Buffer
+	buf.WriteByte(blobCodecZstd)
+	enc, err := zstd.NewWriter(&buf)
+	require.NoError(t, err)
+	_, err = enc.Write(make([]byte, 1<<20))
+	require.NoError(t, err)
+	require.NoError(t, enc.Close())
+
+	var hdr zstd.Header
+	require.NoError(t, hdr.Decode(buf.Bytes()[1:]))
+	require.False(t, hdr.HasFCS)
+
+	// Stopped by the decoder, not by the length check after it.
+	_, err = decodeBlobPayload(buf.Bytes(), 4096)
+	require.ErrorIs(t, err, zstd.ErrDecoderSizeExceeded)
+
+	got, err := decodeBlobPayload(buf.Bytes(), 1<<20)
+	require.NoError(t, err)
+	assert.Len(t, got, 1<<20)
+}
+
+func Test_encodeBlobPayload_DeclaresContentSize(t *testing.T) {
+	framed, err := encodeBlobPayload(make([]byte, 4096))
+	require.NoError(t, err)
+	require.Equal(t, blobCodecZstd, framed[0])
+	var hdr zstd.Header
+	require.NoError(t, hdr.Decode(framed[1:]))
+	require.True(t, hdr.HasFCS)
+	assert.Equal(t, uint64(4096), hdr.FrameContentSize)
+}
+
 func Test_encodeBlobPayload_RejectsOversized(t *testing.T) {
 	_, err := encodeBlobPayload(make([]byte, maxDecompressedBlobPayloadBytes+1))
 	require.ErrorContains(t, err, "blob payload too large")

llo/dev/v31/blobmemo.go

@@ -40,6 +40,13 @@ const maxMemoizedPayloads = ocrtypes.MaxOracles * maxObservationBlobHandles
 // The memo enforces its own budget (maxMemoizedPayloadBytes) and simply declines
 // to store beyond it. Declining is safe because memoization is an optimization,
 // and a miss decodes and costs exactly what a hit would have.
+//
+// Only one sequence number is held at a time. If calls for two sequence numbers
+// overlap, for example the leader validating observations for the next round
+// while StateTransition still runs, each round call discards the other's
+// entries and the superseded handle misses on every get and drops every put.
+// That degrades to running without the memo and bound in the observation
+// decompression budget in decodeObservation, not the memo.
 type blobPayloadCache struct {
 	mu    sync.Mutex
 	seqNr uint64

llo/dev/v31/blobpump.go

@@ -63,6 +63,10 @@ const (
 	// roughly one blob per round, so the unexpired-blob budget declared to
 	// libocr grows with the lifetime; this keeps that budget sane.
 	MaxBlobLifetimeRounds = 64
+	// MinMaxSnapshotRounds bounds MaxSnapshotRounds from below. A cycle gathers
+	// the snapshot for the round that kicked it, and the snapshot is first taken
+	// by the next round, so a window of one round discards every snapshot.
+	MinMaxSnapshotRounds = 2
 	// BlobBroadcastAttempts is how many times one cycle tries to broadcast the
 	// payload it gathered.
 	BlobBroadcastAttempts = 3
@@ -109,6 +113,9 @@ func ResolveBlobRounds(maxSnapshotRounds, blobLifetimeRounds uint64) (uint64, ui
 	if blobLifetimeRounds == 0 {
 		blobLifetimeRounds = DefaultBlobLifetimeRounds
 	}
+	if maxSnapshotRounds < MinMaxSnapshotRounds {
+		return 0, 0, fmt.Errorf("MaxSnapshotRounds (%d) is below MinMaxSnapshotRounds (%d)", maxSnapshotRounds, MinMaxSnapshotRounds)
+	}
 	if blobLifetimeRounds > MaxBlobLifetimeRounds {
 		return 0, 0, fmt.Errorf("BlobLifetimeRounds (%d) exceeds MaxBlobLifetimeRounds (%d)", blobLifetimeRounds, MaxBlobLifetimeRounds)
 	}
@@ -482,6 +489,18 @@ func (p *blobPump) observe(in pumpInput) (*blobSnapshot, error) {
 	if err := p.ds.Observe(ctx, sv, opts); err != nil {
 		return nil, fmt.Errorf("DataSource.Observe error: %w", err)
 	}
+	// Broadcast only the requested streams. Peers reject an observation over
+	// MaxObservationStreamValuesLength, so unrequested keys from the
+	// DataSource could void this node's whole contribution.
+	if len(sv) > len(in.streams) {
+		requested := make(protocol.StreamValues, len(in.streams))
+		for _, sid := range in.streams {
+			requested[sid] = sv[sid]
+		}
+		p.lggr.Warnw("DataSource returned unrequested streams; dropping them", "seqNr", in.seqNr, "requested", len(in.streams), "returned", len(sv))
+		sv = requested
+	}
+	dropUnobservable(sv, p.lggr, in.seqNr)
 
 	payload, err := marshalStreamValues(sv)
 	if err != nil {
@@ -555,6 +574,27 @@ func (p *blobPump) latestSeqNr() uint64 {
 	return p.input.seqNr
 }
 
+// dropUnobservable unsets every value peers would reject on decode, such as one
+// over MaxDecimalCoefficientBits. Peers reject the whole observation on the
+// first such value, so one bad value from the DataSource would otherwise void
+// every other value this node contributes. The check is the receiver's own
+// decode, so both sides apply the same rules.
+func dropUnobservable(sv protocol.StreamValues, lggr logger.Logger, seqNr uint64) {
+	for id, v := range sv {
+		if v == nil {
+			continue
+		}
+		pb, err := protocol.StreamValueToProto(v)
+		if err == nil {
+			_, err = streamValueFromProtoAllowNil(pb)
+		}
+		if err != nil {
+			lggr.Warnw("Dropping stream value peers would reject", "streamID", id, "err", err, "seqNr", seqNr)
+			sv[id] = nil
+		}
+	}
+}
+
 // marshalStreamValues serializes stream values into the stream-values-only
 // proto that is carried by a blob, framed by encodeBlobPayload (which
 // compresses it when that shrinks the payload). Returns nil when nothing was

llo/dev/v31/blobpump_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"math"
+	"math/big"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -12,6 +13,7 @@ import (
 	"github.com/shopspring/decimal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
@@ -92,6 +94,45 @@ func Test_blobPump_TakeKicksNextCycle(t *testing.T) {
 	require.Eventually(t, func() bool { return p.Cycles() >= 2 }, tests.WaitTimeout(t), 10*time.Millisecond)
 }
 
+func Test_blobPump_DropsUnrequestedStreams(t *testing.T) {
+	ds := mockDS()
+	ds.vals[999] = protocol.ToDecimal(decimal.NewFromInt(9))
+	p := testPump(t, ds, newFakeBroadcaster(), time.Minute)
+
+	p.SetInput(pumpInputFor(2))
+	p.Take(t.Context(), 2)
+	require.Eventually(t, func() bool { return p.Cycles() >= 1 }, tests.WaitTimeout(t), 10*time.Millisecond)
+
+	p.SetInput(pumpInputFor(3))
+	snap, reason := p.Take(t.Context(), 3)
+	require.NotNil(t, snap, "reason: %s", reason)
+	require.Equal(t, 1, snap.streamCount)
+}
+
+func Test_dropUnobservable(t *testing.T) {
+	over := new(big.Int).Lsh(big.NewInt(1), protocol.MaxDecimalCoefficientBits+1)
+	sv := protocol.StreamValues{
+		1: protocol.ToDecimal(decimal.NewFromInt(7)),
+		2: protocol.ToDecimal(decimal.NewFromBigInt(over, 0)),
+		3: nil,
+	}
+	dropUnobservable(sv, logger.Test(t), 1)
+	assert.Equal(t, protocol.ToDecimal(decimal.NewFromInt(7)), sv[1])
+	assert.Nil(t, sv[2])
+	assert.Nil(t, sv[3])
+
+	// What is left decodes on the receiving side.
+	payload, err := marshalStreamValues(sv)
+	require.NoError(t, err)
+	raw, err := decodeBlobPayload(payload, maxDecompressedBlobPayloadBytes)
+	require.NoError(t, err)
+	chunk := &protocol.LLOObservationProto{}
+	require.NoError(t, proto.Unmarshal(raw, chunk))
+	require.Len(t, chunk.StreamValues, 1)
+	_, err = streamValueFromProtoAllowNil(chunk.StreamValues[1])
+	require.NoError(t, err)
+}
+
 // Test_blobPump_TakeIsSingleUse guards against two rounds referencing the same
 // blob handle.
 func Test_blobPump_TakeIsSingleUse(t *testing.T) {
@@ -696,6 +737,11 @@ func Test_ResolveBlobRounds(t *testing.T) {
 		assert.Equal(t, uint64(DefaultBlobLifetimeRounds), life)
 	})
 
+	t.Run("rejects a snapshot window below the minimum", func(t *testing.T) {
+		_, _, err := ResolveBlobRounds(1, 0)
+		require.ErrorContains(t, err, "below MinMaxSnapshotRounds")
+	})
+
 	t.Run("rejects a lifetime past the maximum", func(t *testing.T) {
 		_, _, err := ResolveBlobRounds(0, MaxBlobLifetimeRounds+1)
 		require.ErrorContains(t, err, "exceeds MaxBlobLifetimeRounds")
@@ -707,7 +753,7 @@ func Test_ResolveBlobRounds(t *testing.T) {
 	})
 
 	t.Run("rejects a lifetime shorter than the margin", func(t *testing.T) {
-		_, _, err := ResolveBlobRounds(1, 1)
+		_, _, err := ResolveBlobRounds(MinMaxSnapshotRounds, 1)
 		require.ErrorContains(t, err, "rounds of fetch margin")
 	})