SPOR recommendations

Makefile

@@ -13,7 +13,7 @@ test:
 
 .PHONY: test-ci
 test-ci: testdb
-	go test ./... -covermode=atomic -coverpkg=./... -coverprofile=./coverage.txt -json | tee output.txt
+	go test ./... -covermode=atomic -race -coverpkg=./... -coverprofile=./coverage.txt -json | tee output.txt
 
 .PHONY: lint
 lint:

llo/dev/v31/backfill.go

@@ -59,5 +59,17 @@ func selectBackfillCandidate(defs llotypes.ChannelDefinitions, validAfter map[ll
 	if !found {
 		return 0, 0, protocol.HistoryBackfillOpts{}, false
 	}
+	// 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
+	}
 	return bestNanos, bestRaw, o, true
 }

llo/dev/v31/blobcompress.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/klauspost/compress/zstd"
 
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types"
+
 	protocol "github.com/smartcontractkit/chainlink-data-streams/llo/protocol"
 )
 
@@ -24,6 +26,15 @@ const (
 // a huge allocation (zstd bomb).
 const maxDecompressedBlobPayloadBytes = protocol.MaxDecompressedObservationLength
 
+// MaxBlobPayloadBytes is the size of a broadcast blob payload libocr accepts,
+// and is the value the factory declares as MaxBlobPayloadBytes. It bounds the
+// payload as framed for broadcast (codec byte plus body), which is a different
+// quantity from maxDecompressedBlobPayloadBytes: the latter is the anti-bomb
+// bound applied to the decompressed bytes on the read side, and is deliberately
+// looser. Enforcing this one on the write side is what stops the pump from
+// broadcasting a blob every peer's libocr would reject.
+const MaxBlobPayloadBytes = ocr3_1types.MaxMaxBlobPayloadBytes
+
 // zstd Encoder/Decoder are safe for concurrent use via EncodeAll/DecodeAll and
 // are expensive to build, so a single pair is shared process-wide. Built lazily
 // so a construction failure surfaces at the call site rather than in init.
@@ -74,12 +85,18 @@ func encodeBlobPayload(raw []byte) ([]byte, error) {
 		return nil, err
 	}
 	compressed := c.encoder.EncodeAll(raw, []byte{blobCodecZstd})
-	if len(compressed) < len(raw)+1 {
-		return compressed, nil
+	out := compressed
+	if len(compressed) >= len(raw)+1 {
+		out = make([]byte, 0, len(raw)+1)
+		out = append(out, blobCodecRaw)
+		out = append(out, raw...)
+	}
+	// The framed payload is what is broadcast, has to fit the declared limit.
+	// A payload that compresses poorly can pass the check above and still land over it.
+	if len(out) > MaxBlobPayloadBytes {
+		return nil, fmt.Errorf("framed blob payload too large: %d > %d bytes", len(out), MaxBlobPayloadBytes)
 	}
-	out := make([]byte, 0, len(raw)+1)
-	out = append(out, blobCodecRaw)
-	return append(out, raw...), nil
+	return out, nil
 }
 
 // decodeBlobPayload reverses encodeBlobPayload. The payload is untrusted, so

llo/dev/v31/blobcompress_fuzz_test.go

@@ -0,0 +1,33 @@
+package llo
+
+import (
+	"testing"
+)
+
+// FuzzDecodeBlobPayload feeds arbitrary bytes through the blob payload decoder.
+// Blob payloads are attacker-controlled, so the contract is that any input
+// either returns bytes within the caller's budget or errors. Never a panic,
+// and never more bytes than the budget allows (a zstd bomb).
+func FuzzDecodeBlobPayload(f *testing.F) {
+	raw := []byte("stream values would go here")
+	framed, err := encodeBlobPayload(raw)
+	if err != nil {
+		f.Fatal(err)
+	}
+	f.Add(framed)
+	f.Add(append([]byte{blobCodecRaw}, raw...))
+	f.Add([]byte{blobCodecZstd})
+	f.Add([]byte{})
+
+	f.Fuzz(func(t *testing.T, payload []byte) {
+		for _, budget := range []int{0, 1, 1 << 10, maxDecompressedBlobPayloadBytes} {
+			out, err := decodeBlobPayload(payload, budget)
+			if err != nil {
+				continue
+			}
+			if len(out) > budget {
+				t.Fatalf("decoded %d bytes against a budget of %d", len(out), budget)
+			}
+		}
+	})
+}

llo/dev/v31/blobmemo.go

@@ -0,0 +1,126 @@
+package llo
+
+import (
+	"sync"
+
+	protocol "github.com/smartcontractkit/chainlink-data-streams/llo/protocol"
+
+	ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
+)
+
+// maxMemoizedPayloadBytes bounds the decompressed bytes one round's memo
+// may hold across all of its entries.The budget is a small multiple of what
+// one observation may carry, which covers honest traffic while keeping the
+// worst case independent of N.
+const maxMemoizedPayloadBytes = 4 * maxObservationDecompressedBytes
+
+// maxMemoizedPayloads bounds the entry count, which the byte budget alone does
+// not: an empty payload costs no bytes. No round can legitimately reference more
+// handles than every oracle naming the most it is allowed.
+const maxMemoizedPayloads = ocrtypes.MaxOracles * maxObservationBlobHandles
+
+// blobPayloadCache memoizes the stream values decoded from blob payloads within
+// one sequence number. ValidateObservation and StateTransition decode the same
+// observations in the same round, and every FetchBlob re-verifies the blob's
+// certificate and re-reads its payload before the plugin decompresses and
+// unmarshals it again, so the second decode is entirely repeated work.
+//
+// Entries are keyed by the marshaled blob handle and scoped to a single
+// sequence number. That scope is what keeps the memo consistent with the blob
+// transport, which refuses a handle that has expired as of the round's sequence
+// number: a hit can never resurrect a blob the round itself would have
+// rejected. What one round can hold is bounded by what its observations may
+// reference, at most maxObservationBlobHandles handles per observation, each
+// contributing at most maxObservationDecompressedBytes, and the whole map is
+// dropped when the sequence number advances.
+//
+// Decoded stream values are treated as immutable: a hit copies the map entries
+// into the observation rather than handing out the memoized map.
+//
+// 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.
+type blobPayloadCache struct {
+	mu    sync.Mutex
+	seqNr uint64
+	// bytes is the sum of entries' sizes, tracked so the budget does not have
+	// to walk the map on every write.
+	bytes   int
+	entries map[string]blobPayloadEntry
+}
+
+// blobPayloadEntry is one memoized blob payload. size is the decompressed byte
+// count, memoized alongside the values because it is charged against the
+// observation's decompression budget: a hit must consume exactly what the
+// original decode consumed, or the same observation would decode differently on
+// a hit than on a miss.
+type blobPayloadEntry struct {
+	values protocol.StreamValues
+	size   int
+}
+
+func newBlobPayloadCache() *blobPayloadCache {
+	return &blobPayloadCache{}
+}
+
+// round returns the memo scoped to seqNr, discarding whatever was held for an
+// earlier one. Returns nil for a nil cache, which disables memoization.
+func (c *blobPayloadCache) round(seqNr uint64) *roundBlobPayloads {
+	if c == nil {
+		return nil
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.seqNr != seqNr || c.entries == nil {
+		c.seqNr = seqNr
+		c.entries = make(map[string]blobPayloadEntry)
+		c.bytes = 0
+	}
+	return &roundBlobPayloads{cache: c, seqNr: seqNr}
+}
+
+// roundBlobPayloads is a handle on the memo for one sequence number. Reads and
+// writes through a handle whose round has been superseded are dropped, so a
+// round can never see another round's payloads.
+type roundBlobPayloads struct {
+	cache *blobPayloadCache
+	seqNr uint64
+}
+
+func (r *roundBlobPayloads) get(handle []byte) (blobPayloadEntry, bool) {
+	if r == nil {
+		return blobPayloadEntry{}, false
+	}
+	r.cache.mu.Lock()
+	defer r.cache.mu.Unlock()
+	if r.cache.seqNr != r.seqNr {
+		return blobPayloadEntry{}, false
+	}
+	entry, ok := r.cache.entries[string(handle)]
+	return entry, ok
+}
+
+func (r *roundBlobPayloads) put(handle []byte, entry blobPayloadEntry) {
+	if r == nil {
+		return
+	}
+	r.cache.mu.Lock()
+	defer r.cache.mu.Unlock()
+	if r.cache.seqNr != r.seqNr {
+		return
+	}
+	key := string(handle)
+	prev, replacing := r.cache.entries[key]
+	if !replacing && len(r.cache.entries) >= maxMemoizedPayloads {
+		return
+	}
+	bytes := r.cache.bytes + entry.size
+	if replacing {
+		bytes -= prev.size
+	}
+	if bytes > maxMemoizedPayloadBytes {
+		return
+	}
+	r.cache.entries[key] = entry
+	r.cache.bytes = bytes
+}