Fix non-determinism in Reports()

llo/protocol/opts_cache.go

@@ -69,11 +69,7 @@ func (c *OptsCache) Set(channelID llotypes.ChannelID, raw llotypes.ChannelOpts)
 		return
 	}
 	c.raw[channelID] = raw
-	for key := range c.decoded {
-		if key.channelID == channelID {
-			delete(c.decoded, key)
-		}
-	}
+	c.invalidateDecoded(channelID)
 }
 
 // Len returns the number of channels in the cache.
@@ -93,13 +89,47 @@ func (c *OptsCache) Remove(channelID llotypes.ChannelID) {
 	}
 
 	delete(c.raw, channelID)
+	c.invalidateDecoded(channelID)
+}
+
+// invalidateDecoded drops every decoded value for a channel. Callers must hold c.mu.
+func (c *OptsCache) invalidateDecoded(channelID llotypes.ChannelID) {
 	for key := range c.decoded {
 		if key.channelID == channelID {
 			delete(c.decoded, key)
 		}
 	}
 }
 
+// SyncTo makes the cache's contents match channelDefinitions exactly.
+//
+// Unlike ResetTo it compares raw opts by content, so already decoded values
+// survive for channels whose opts are unchanged. That makes it cheap enough to
+// call on every round in order to guarantee the cache agrees with a given set
+// of channel definitions, rather than relying on incremental Set/Remove calls
+// having kept it in step.
+func (c *OptsCache) SyncTo(channelDefinitions llotypes.ChannelDefinitions) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for channelID, cd := range channelDefinitions {
+		if existing, ok := c.raw[channelID]; ok && bytes.Equal(existing, cd.Opts) {
+			continue
+		}
+		c.raw[channelID] = cd.Opts
+		c.invalidateDecoded(channelID)
+	}
+
+	// Drop channels that are no longer defined. Deleting during a range over
+	// the same map is safe in Go.
+	for channelID := range c.raw {
+		if _, ok := channelDefinitions[channelID]; !ok {
+			delete(c.raw, channelID)
+			c.invalidateDecoded(channelID)
+		}
+	}
+}
+
 // ResetTo resets the cache to the given channel definitions.
 func (c *OptsCache) ResetTo(channelDefinitions llotypes.ChannelDefinitions) {
 	c.mu.Lock()

llo/protocol/opts_cache_test.go

@@ -1,6 +1,7 @@
 package protocol
 
 import (
+	"reflect"
 	"testing"
 
 	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
@@ -289,3 +290,86 @@ func TestOptsCache_ChannelDefinitionWorkflow(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "ch1", r1Again.FeedID)
 }
+
+func TestOptsCache_SyncTo(t *testing.T) {
+	defsFor := func(opts map[llotypes.ChannelID]string) llotypes.ChannelDefinitions {
+		defs := make(llotypes.ChannelDefinitions, len(opts))
+		for cid, o := range opts {
+			defs[cid] = llotypes.ChannelDefinition{Opts: llotypes.ChannelOpts(o)}
+		}
+		return defs
+	}
+
+	t.Run("adds channels that are not yet present", func(t *testing.T) {
+		cache := NewOptsCache()
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{
+			1: `{"feedID":"ch1"}`,
+			2: `{"feedID":"ch2"}`,
+		}))
+
+		assert.Equal(t, 2, cache.Len())
+		r1, err := GetOpts[testOptsA](cache, 1)
+		require.NoError(t, err)
+		assert.Equal(t, "ch1", r1.FeedID)
+	})
+
+	t.Run("drops channels that are no longer defined", func(t *testing.T) {
+		cache := NewOptsCache()
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"ch1"}`, 2: `{"feedID":"ch2"}`}))
+		_, err := GetOpts[testOptsA](cache, 2)
+		require.NoError(t, err)
+
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"ch1"}`}))
+
+		assert.Equal(t, 1, cache.Len())
+		_, err = GetOpts[testOptsA](cache, 2)
+		require.Error(t, err, "channel 2 should be gone")
+	})
+
+	t.Run("picks up changed opts for an existing channel", func(t *testing.T) {
+		cache := NewOptsCache()
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"before"}`}))
+		r, err := GetOpts[testOptsA](cache, 1)
+		require.NoError(t, err)
+		require.Equal(t, "before", r.FeedID)
+
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"after"}`}))
+
+		r, err = GetOpts[testOptsA](cache, 1)
+		require.NoError(t, err)
+		assert.Equal(t, "after", r.FeedID, "SyncTo must compare opts by content, not just by channel ID")
+	})
+
+	// The point of SyncTo over ResetTo: syncing to unchanged definitions must
+	// not throw away decoded values, otherwise every round re-decodes.
+	t.Run("preserves decoded values for unchanged opts", func(t *testing.T) {
+		defs := defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"ch1"}`, 2: `{"feedID":"ch2"}`})
+		cache := NewOptsCache()
+		cache.SyncTo(defs)
+		_, err := GetOpts[testOptsA](cache, 1)
+		require.NoError(t, err)
+		_, err = GetOpts[testOptsA](cache, 2)
+		require.NoError(t, err)
+		require.Len(t, cache.decoded, 2)
+
+		cache.SyncTo(defs)
+		assert.Len(t, cache.decoded, 2, "decoded values should survive a sync to identical definitions")
+
+		// A changed channel invalidates only its own decoded values.
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"ch1"}`, 2: `{"feedID":"ch2-new"}`}))
+		assert.Len(t, cache.decoded, 1)
+		_, ok := cache.decoded[optsCacheKey{channelID: 1, optsType: reflect.TypeFor[testOptsA]()}]
+		assert.True(t, ok, "channel 1 was unchanged and should still be decoded")
+	})
+
+	t.Run("syncing to empty definitions clears the cache", func(t *testing.T) {
+		cache := NewOptsCache()
+		cache.SyncTo(defsFor(map[llotypes.ChannelID]string{1: `{"feedID":"ch1"}`}))
+		_, err := GetOpts[testOptsA](cache, 1)
+		require.NoError(t, err)
+
+		cache.SyncTo(nil)
+		assert.Equal(t, 0, cache.Len())
+		assert.Empty(t, cache.decoded)
+	})
+}

llo/v30/plugin.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"sync"
 	"time"
 
 	"github.com/smartcontractkit/libocr/quorumhelper"
@@ -177,6 +178,8 @@ func (f *PluginFactory) NewReportingPlugin(ctx context.Context, cfg ocr3types.Re
 			f.ReportTelemetryCh,
 			f.DonID,
 			protocol.NewOptsCache(),
+			sync.Mutex{},
+			protocol.NewOptsCache(),
 			cfg.MaxDurationObservation,
 			offchainConfig.ProtocolVersion,
 			offchainConfig.DefaultMinReportIntervalNanoseconds,
@@ -214,6 +217,16 @@ type Plugin struct {
 	DonID                            uint32
 	OptsCache                        *protocol.OptsCache // must be non-nil; set by NewReportingPlugin or by tests that exercise Outcome/Reports
 
+	// reportsOptsCache memoizes opts decoding for Reports() across rounds. It is
+	// deliberately separate from OptsCache, which Outcome() owns and mutates
+	// from its own goroutine, and it is synced to the committed outcome's
+	// channel definitions on every call so that it cannot go stale. reportsMu
+	// guards it and serializes Reports(), which libocr may invoke concurrently
+	// for different sequence numbers. Lazily initialized, so a hand-built
+	// Plugin needs no extra setup.
+	reportsMu        sync.Mutex
+	reportsOptsCache *protocol.OptsCache
+
 	// From ReportingPluginConfig
 	MaxDurationObservation time.Duration
 

llo/v30/plugin_reports.go

@@ -23,6 +23,28 @@ func (p *Plugin) reports(ctx context.Context, seqNr uint64, rawOutcome ocr3types
 		return nil, fmt.Errorf("error unmarshalling outcome: %w", err)
 	}
 
+	// Decode channel opts from this outcome's channel definitions rather than
+	// from p.OptsCache.
+	//
+	// p.OptsCache is node-local state populated only by Outcome(), but libocr
+	// can call Reports() for a committed outcome this node never computed
+	// itself. p.OptsCache is then empty, opts-dependent report codecs fail, and the
+	// channel is skipped, so it emits a different report set than its peers.
+	//
+	// reportsOptsCache is reserved for Reports() and synced to this outcome's
+	// channel definitions by content, which keeps Reports() a pure function of
+	// (seqNr, outcome) while still memoizing the JSON decoding across rounds.
+	// The lock is held for the remainder of the call: two concurrent Reports()
+	// calls syncing the same cache to different outcomes would otherwise read
+	// each other's channel definitions.
+	p.reportsMu.Lock()
+	defer p.reportsMu.Unlock()
+	if p.reportsOptsCache == nil {
+		p.reportsOptsCache = protocol.NewOptsCache()
+	}
+	optsCache := p.reportsOptsCache
+	optsCache.SyncTo(outcome.ChannelDefinitions)
+
 	rwis := []ocr3types.ReportPlus[llotypes.ReportInfo]{}
 
 	if outcome.LifeCycleStage == protocol.LifeCycleStageRetired {
@@ -48,7 +70,7 @@ func (p *Plugin) reports(ctx context.Context, seqNr uint64, rawOutcome ocr3types
 		})
 	}
 
-	reportableChannels, unreportableChannels := outcome.ReportableChannels(p.ProtocolVersion, p.DefaultMinReportIntervalNanoseconds, p.OptsCache)
+	reportableChannels, unreportableChannels := outcome.ReportableChannels(p.ProtocolVersion, p.DefaultMinReportIntervalNanoseconds, optsCache)
 	if p.Config.VerboseLogging {
 		p.Logger.Debugw("Reportable channels", "lifeCycleStage", outcome.LifeCycleStage, "reportableChannels", reportableChannels, "unreportableChannels", unreportableChannels, "stage", "Report", "seqNr", seqNr)
 	}
@@ -105,7 +127,7 @@ func (p *Plugin) reports(ctx context.Context, seqNr uint64, rawOutcome ocr3types
 				p.Logger.Warnw("Error encoding report", "lifeCycleStage", outcome.LifeCycleStage, "reportFormat", targetCD.ReportFormat, "err", fmt.Errorf("codec missing for ReportFormat=%q", targetCD.ReportFormat), "channelID", cid, "stage", "Report", "seqNr", seqNr)
 				continue
 			}
-			encoded, err := codec.Encode(reportForEncode, targetCD, p.OptsCache)
+			encoded, err := codec.Encode(reportForEncode, targetCD, optsCache)
 			if err != nil {
 				p.Logger.Warnw("Error encoding report", "lifeCycleStage", outcome.LifeCycleStage, "reportFormat", targetCD.ReportFormat, "err", err, "channelID", cid, "stage", "Report", "seqNr", seqNr)
 				continue
@@ -141,7 +163,7 @@ func (p *Plugin) reports(ctx context.Context, seqNr uint64, rawOutcome ocr3types
 			p.Logger.Debugw("Emitting report", "lifeCycleStage", outcome.LifeCycleStage, "channelID", cid, "report", report, "stage", "Report", "seqNr", seqNr)
 		}
 
-		encoded, err := p.encodeReport(report, cd)
+		encoded, err := p.encodeReport(report, cd, optsCache)
 		if err != nil {
 			p.Logger.Warnw("Error encoding report", "lifeCycleStage", outcome.LifeCycleStage, "reportFormat", cd.ReportFormat, "err", err, "channelID", cid, "stage", "Report", "seqNr", seqNr)
 			continue
@@ -164,13 +186,13 @@ func (p *Plugin) reports(ctx context.Context, seqNr uint64, rawOutcome ocr3types
 	return rwis, nil
 }
 
-func (p *Plugin) encodeReport(r protocol.Report, cd llotypes.ChannelDefinition) (types.Report, error) {
+func (p *Plugin) encodeReport(r protocol.Report, cd llotypes.ChannelDefinition, optsCache *protocol.OptsCache) (types.Report, error) {
 	codec, exists := p.ReportCodecs[cd.ReportFormat]
 	if !exists {
 		return nil, fmt.Errorf("codec missing for ReportFormat=%q", cd.ReportFormat)
 	}
 	p.captureReportTelemetry(r, cd)
-	return codec.Encode(r, cd, p.OptsCache)
+	return codec.Encode(r, cd, optsCache)
 }
 
 func (p *Plugin) captureReportTelemetry(r protocol.Report, cd llotypes.ChannelDefinition) {