feat(llo/v31): skip observation/aggregation for recently-reported channels

llo/dev/v31/blobpump_test.go

@@ -222,8 +222,8 @@ func Test_observableStreams(t *testing.T) {
 		2: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}},
 		3: {ReportFormat: llotypes.ReportFormatJSON, Tombstone: true, Streams: []llotypes.Stream{{StreamID: 102, Aggregator: llotypes.AggregatorMedian}}},
 	}}
-	require.ElementsMatch(t, []llotypes.StreamID{100}, observableStreams(state))
-	require.Empty(t, observableStreams(&kvState{}))
+	require.ElementsMatch(t, []llotypes.StreamID{100}, observableStreams(state, 0, 0))
+	require.Empty(t, observableStreams(&kvState{}, 0, 0))
 }
 
 // Test_blobPump_DisabledIsInert covers hosts that run the plugin without blob

llo/dev/v31/doc.go

@@ -27,8 +27,9 @@
 // how many channels and streams exist:
 //
 //   - r/agg holds the per-round ("hot") state — observation timestamp,
-//     validAfter watermarks, per-channel reportability, and carry-forward
-//     timestamped aggregates — and is rewritten every round.
+//     validAfter watermarks, per-channel reportability, the observation
+//     schedule, and carry-forward timestamped aggregates — and is rewritten
+//     every round.
 //   - c/defs holds every channel definition and is rewritten only when the
 //     definitions change; c/seqnr records the sequence number of that write.
 //   - c/lifecycle holds the lifecycle stage and is written only on change.

llo/dev/v31/factory.go

@@ -94,23 +94,24 @@ func (f *PluginFactory) NewReportingPlugin(ctx context.Context, cfg ocr3types.Re
 	}
 
 	p := &Plugin{
-		Config:                              f.Config,
-		PredecessorConfigDigest:             onchainConfig.PredecessorConfigDigest,
-		ConfigDigest:                        cfg.ConfigDigest,
-		PredecessorRetirementReportCache:    f.PredecessorRetirementReportCache,
-		ShouldRetireCache:                   f.ShouldRetireCache,
-		ChannelDefinitionCache:              f.ChannelDefinitionCache,
-		DataSource:                          f.DataSource,
-		Logger:                              l,
-		N:                                   cfg.N,
-		F:                                   cfg.F,
-		RetirementReportCodec:               f.RetirementReportCodec,
-		ReportCodecs:                        f.ReportCodecs,
-		DonID:                               f.DonID,
-		OutcomeTelemetryCh:                  f.OutcomeTelemetryCh,
-		ReportTelemetryCh:                   f.ReportTelemetryCh,
-		ProtocolVersion:                     offchainConfig.ProtocolVersion,
-		DefaultMinReportIntervalNanoseconds: offchainConfig.DefaultMinReportIntervalNanoseconds,
+		Config:                                   f.Config,
+		PredecessorConfigDigest:                  onchainConfig.PredecessorConfigDigest,
+		ConfigDigest:                             cfg.ConfigDigest,
+		PredecessorRetirementReportCache:         f.PredecessorRetirementReportCache,
+		ShouldRetireCache:                        f.ShouldRetireCache,
+		ChannelDefinitionCache:                   f.ChannelDefinitionCache,
+		DataSource:                               f.DataSource,
+		Logger:                                   l,
+		N:                                        cfg.N,
+		F:                                        cfg.F,
+		RetirementReportCodec:                    f.RetirementReportCodec,
+		ReportCodecs:                             f.ReportCodecs,
+		DonID:                                    f.DonID,
+		OutcomeTelemetryCh:                       f.OutcomeTelemetryCh,
+		ReportTelemetryCh:                        f.ReportTelemetryCh,
+		ProtocolVersion:                          offchainConfig.ProtocolVersion,
+		DefaultMinReportIntervalNanoseconds:      offchainConfig.DefaultMinReportIntervalNanoseconds,
+		DefaultMinObservationIntervalNanoseconds: offchainConfig.DefaultMinObservationIntervalNanoseconds,
 	}
 
 	// Definitions and the opts decoded from them are cached together, as one

llo/dev/v31/flow_test.go

@@ -270,6 +270,90 @@ func Test_StateTransition_ChannelRemoval(t *testing.T) {
 	require.NotContains(t, hot.reportedLastRound, llotypes.ChannelID(1))
 }
 
+// Promotion replaces validAfter wholesale from the predecessor's retirement
+// report so the handover is gapless. The observation schedule must be cleared to
+// match: a schedule carried over from staging can leave a channel not due on the
+// promotion round, and a channel that is not aggregated has no values and so
+// cannot report, reopening the gap promotion exists to avoid.
+func Test_StateTransition_PromotionClearsObservationSchedule(t *testing.T) {
+	ctx := tests.Context(t)
+	p := testPlugin(t)
+	p.DefaultMinReportIntervalNanoseconds = 5000
+	p.DefaultMinObservationIntervalNanoseconds = 5000
+	predecessor := ocrtypes.ConfigDigest{0xAB}
+	p.PredecessorConfigDigest = &predecessor
+	p.PredecessorRetirementReportCache = &mockPredecessorRetirementReportCache{
+		// Old watermark: the channel is immediately report-due on promotion.
+		report: protocol.RetirementReport{ValidAfterNanoseconds: map[llotypes.ChannelID]uint64{1: 1}},
+	}
+	channelDef := llotypes.ChannelDefinition{
+		ReportFormat: llotypes.ReportFormatJSON,
+		Streams:      []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}},
+	}
+	kv := newMemKV()
+
+	_, err := p.StateTransition(ctx, 1, ocrtypes.AttributedQuery{}, []ocrtypes.AttributedObservation{ao(0, nil), ao(1, nil), ao(2, nil)}, kv, testBlobs)
+	require.NoError(t, err)
+	_, err = p.StateTransition(ctx, 2, ocrtypes.AttributedQuery{}, addChannelRound(t, 1_000, 1, channelDef), kv, testBlobs)
+	require.NoError(t, err)
+
+	valued := func(ts uint64, v int64) []ocrtypes.AttributedObservation {
+		obs := Observation{
+			UnixTimestampNanoseconds: ts,
+			StreamValues:             protocol.StreamValues{100: protocol.ToDecimal(decimal.NewFromInt(v))},
+		}
+		aos := make([]ocrtypes.AttributedObservation, 0, 4)
+		for i := 0; i < 4; i++ {
+			aos = append(aos, ao(i, mustEncodeObs(t, obs)))
+		}
+		return aos
+	}
+
+	// Run the staging instance until it has reported and so acquired a schedule
+	// slot some way in the future.
+	seqNr, ts := uint64(3), uint64(3_000)
+	for range 6 {
+		_, err = p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valued(ts, 10), kv, testBlobs)
+		require.NoError(t, err)
+		seqNr++
+		ts += 3_000
+		if storedObservationDue(t, kv, 1) != 0 {
+			break
+		}
+	}
+	scheduled := storedObservationDue(t, kv, 1)
+	require.NotZero(t, scheduled, "staging must acquire a schedule slot for this test to mean anything")
+
+	// Promote while that slot is still in the future, which is the case the
+	// carried-over schedule would break.
+	ts = scheduled - 1_000
+	require.Greater(t, scheduled, ts)
+
+	// Promote. validAfter is reseeded from the retirement report, so the channel
+	// is report-due immediately; it must also be aggregated immediately.
+	promo := Observation{UnixTimestampNanoseconds: ts, AttestedPredecessorRetirement: []byte("attested")}
+	promoAOs := make([]ocrtypes.AttributedObservation, 0, 4)
+	for i := 0; i < 4; i++ {
+		promoAOs = append(promoAOs, ao(i, mustEncodeObs(t, promo)))
+	}
+	_, err = p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, promoAOs, kv, testBlobs)
+	require.NoError(t, err)
+	seqNr++
+	ts += 3_000
+
+	require.Equal(t, string(protocol.LifeCycleStageProduction), string(kv.m[string(keyLifecycle)]))
+	require.Zero(t, storedObservationDue(t, kv, 1),
+		"promotion must clear the schedule so the channel is due immediately")
+
+	// The next round must aggregate it, which is what lets it report.
+	prec, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valued(ts, 20), kv, testBlobs)
+	require.NoError(t, err)
+	decoded, err := decodePrecursor(prec)
+	require.NoError(t, err)
+	require.NotNil(t, decoded.StreamAggregates[100][llotypes.AggregatorMedian],
+		"a promoted channel must be aggregated rather than waiting for a staging schedule slot")
+}
+
 func Test_StateTransition_Promotion(t *testing.T) {
 	ctx := tests.Context(t)
 	p := testPlugin(t)

llo/dev/v31/history_flow_test.go

@@ -103,6 +103,86 @@ func Test_History_Warmup(t *testing.T) {
 	}
 }
 
+// Test_History_SurvivesObservationSkip covers the interaction between the
+// observation interval and stream history. A channel reading history is not
+// exempt from the skip, so its window is sampled at its report cadence: it must
+// still gain exactly one record per cycle it is aggregated, gain none on the
+// rounds it is skipped, and stay readable across the skip.
+func Test_History_SurvivesObservationSkip(t *testing.T) {
+	ctx := tests.Context(t)
+	const depth = 3
+	const interval = 50_000
+	expression := fmt.Sprintf("Count(History(s100, %d))", depth)
+
+	p := historyPlugin(t, expression)
+	p.DefaultMinReportIntervalNanoseconds = interval
+	p.DefaultMinObservationIntervalNanoseconds = interval
+	kv := newMemKV()
+	bootstrapHistoryChannel(t, p, kv, expression)
+
+	key := histKey{streamID: 100, aggregator: llotypes.AggregatorMedian}
+	seqNr := uint64(3)
+	ts := uint64(10_000)
+
+	// Warm up and run to the first report. The channel has never reported, so it
+	// has no schedule slot and is aggregated - and appends - every round.
+	var reportedAt uint64
+	lastLen := 0
+	for range 20 {
+		_, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, ts, 1), kv, testBlobs)
+		require.NoError(t, err)
+		seqNr++
+
+		grown := readHistory(t, kv, key.streamID, key.aggregator).Len()
+		require.Greater(t, grown, lastLen, "an unscheduled channel appends every round")
+		lastLen = grown
+
+		if reportedFlag(t, kv, 1) {
+			reportedAt = ts
+			break
+		}
+		ts += 10_000
+	}
+	require.NotZero(t, reportedAt, "the channel must report before a skip window can be exercised")
+	require.Zero(t, storedObservationDue(t, kv, 1), "not scheduled until it has reported")
+
+	// The next round picks up that report and schedules the channel forward.
+	ts += 10_000
+	_, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, ts, 99), kv, testBlobs)
+	require.NoError(t, err)
+	seqNr++
+	dueAt := storedObservationDue(t, kv, 1)
+	require.Equal(t, reportedAt+interval, dueAt, "scheduled one interval on from the round that reported")
+
+	// Rounds inside the skip window append nothing.
+	depthAtSkipStart := readHistory(t, kv, key.streamID, key.aggregator).Len()
+	skipRounds := 0
+	for ts+10_000 < dueAt {
+		ts += 10_000
+		skipRounds++
+		_, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, ts, 1), kv, testBlobs)
+		require.NoError(t, err)
+		seqNr++
+		require.Equal(t, depthAtSkipStart, readHistory(t, kv, key.streamID, key.aggregator).Len(),
+			"a skipped round must not append to the window")
+		require.False(t, reportedFlag(t, kv, 1), "a skipped channel does not report")
+	}
+	require.NotZero(t, skipRounds, "the test must actually exercise skipped rounds")
+
+	// The window is still there and still at its required depth: the skip lowers
+	// the sampling rate, it does not tear the window down.
+	stored := readHistory(t, kv, key.streamID, key.aggregator)
+	require.NotNil(t, stored)
+	require.GreaterOrEqual(t, stored.Len(), depth, "the window stays readable across a skip window")
+	require.Equal(t, uint32(depth), stored.RequiredCount())
+
+	// Coming due again appends exactly one more record.
+	_, err = p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, valueRound(t, dueAt+10_000, 7), kv, testBlobs)
+	require.NoError(t, err)
+	require.Equal(t, depthAtSkipStart+1, readHistory(t, kv, key.streamID, key.aggregator).Len(),
+		"the round the channel is due again appends exactly one record")
+}
+
 // Test_History_EvictsAtDepth checks the window stays bounded across many rounds
 // and keeps the newest values.
 func Test_History_EvictsAtDepth(t *testing.T) {
@@ -277,9 +357,10 @@ func storedChannelDefinitions(t *testing.T, kv *memKV) llotypes.ChannelDefinitio
 func storedHotState(t *testing.T, kv *memKV) *kvState {
 	t.Helper()
 	s := &kvState{
-		validAfterNanoseconds: map[llotypes.ChannelID]uint64{},
-		reportedLastRound:     map[llotypes.ChannelID]bool{},
-		carryForward:          map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{},
+		validAfterNanoseconds:     map[llotypes.ChannelID]uint64{},
+		reportedLastRound:         map[llotypes.ChannelID]bool{},
+		observationDueNanoseconds: map[llotypes.ChannelID]uint64{},
+		carryForward:              map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{},
 	}
 	require.NoError(t, readHotState(kv, s))
 	return s
@@ -291,6 +372,13 @@ func storedValidAfter(t *testing.T, kv *memKV, cid llotypes.ChannelID) uint64 {
 	return storedHotState(t, kv).validAfterNanoseconds[cid]
 }
 
+// storedObservationDue returns the persisted observation schedule slot, or 0 if
+// the channel has none (which means it is due).
+func storedObservationDue(t *testing.T, kv *memKV, cid llotypes.ChannelID) uint64 {
+	t.Helper()
+	return storedHotState(t, kv).observationDueNanoseconds[cid]
+}
+
 // reportedFlag returns the reportability decision the last round persisted.
 func reportedFlag(t *testing.T, kv *memKV, cid llotypes.ChannelID) bool {
 	t.Helper()

llo/dev/v31/kv.go

@@ -115,6 +115,11 @@ type kvState struct {
 	// round can advance validAfter faithfully without re-deriving it from
 	// aggregates that are not persisted.
 	reportedLastRound map[llotypes.ChannelID]bool
+	// observationDueNanoseconds is the observation schedule: when each channel
+	// next becomes due for observation and aggregation. A channel with no entry
+	// is due. Distinct from validAfterNanoseconds, which is a report boundary;
+	// see nextObservationDue for why the two must not be conflated.
+	observationDueNanoseconds map[llotypes.ChannelID]uint64
 	// carryForward holds the timestamped aggregates that survive across rounds
 	// (newer-wins monotonicity). Regular aggregates are recomputed fresh every
 	// round and are never persisted.
@@ -148,10 +153,11 @@ func loadKVState(r ocr3_1types.KeyValueStateReader, cache *protocol.ChannelCache
 // StateTransition needs the hot state and must use loadKVState.
 func loadColdKVState(r ocr3_1types.KeyValueStateReader, cache *protocol.ChannelCache) (*kvState, error) {
 	s := &kvState{
-		channelDefinitions:    llotypes.ChannelDefinitions{},
-		validAfterNanoseconds: map[llotypes.ChannelID]uint64{},
-		reportedLastRound:     map[llotypes.ChannelID]bool{},
-		carryForward:          map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{},
+		channelDefinitions:        llotypes.ChannelDefinitions{},
+		validAfterNanoseconds:     map[llotypes.ChannelID]uint64{},
+		reportedLastRound:         map[llotypes.ChannelID]bool{},
+		observationDueNanoseconds: map[llotypes.ChannelID]uint64{},
+		carryForward:              map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue{},
 	}
 
 	lc, err := r.Read(keyLifecycle)
@@ -223,6 +229,9 @@ func readHotState(r ocr3_1types.KeyValueStateReader, s *kvState) error {
 	for _, cid := range pb.ReportableChannelIDs {
 		s.reportedLastRound[cid] = true
 	}
+	for _, d := range pb.ObservationDueNanoseconds {
+		s.observationDueNanoseconds[d.ChannelID] = d.DueAtNanoseconds
+	}
 	for _, sa := range pb.StreamAggregates {
 		sv, err := protocol.UnmarshalProtoStreamValue(sa.StreamValue)
 		if err != nil {
@@ -242,6 +251,36 @@ func readHotState(r ocr3_1types.KeyValueStateReader, s *kvState) error {
 	return nil
 }
 
+// readHotStateForObservation reads the r/agg record and extracts everything the
+// Observation phase needs to decide which channels are due - the observation
+// schedule, the previous observation timestamp, the validAfter watermarks and
+// the reportability flags - while skipping the (potentially large)
+// carry-forward stream aggregates, which only StateTransition uses.
+func readHotStateForObservation(r ocr3_1types.KeyValueStateReader, s *kvState) error {
+	b, err := r.Read(keyHotState)
+	if err != nil {
+		return fmt.Errorf("read hot state: %w", err)
+	}
+	if len(b) == 0 {
+		return nil
+	}
+	pb := &protocol.LLOHotStateProto{}
+	if err := proto.Unmarshal(b, pb); err != nil {
+		return fmt.Errorf("unmarshal hot state: %w", err)
+	}
+	s.observationTimestampNs = pb.ObservationTimestampNanoseconds
+	for _, va := range pb.ValidAfterNanoseconds {
+		s.validAfterNanoseconds[va.ChannelID] = va.ValidAfterNanoseconds
+	}
+	for _, cid := range pb.ReportableChannelIDs {
+		s.reportedLastRound[cid] = true
+	}
+	for _, d := range pb.ObservationDueNanoseconds {
+		s.observationDueNanoseconds[d.ChannelID] = d.DueAtNanoseconds
+	}
+	return nil
+}
+
 // writeLifecycle persists the lifecycle stage.
 func writeLifecycle(w ocr3_1types.KeyValueStateReadWriter, stage llotypes.LifeCycleStage) error {
 	return w.Write(keyLifecycle, []byte(stage))
@@ -281,6 +320,7 @@ func writeHotState(
 	observationTimestampNs uint64,
 	validAfterNanoseconds map[llotypes.ChannelID]uint64,
 	reportable map[llotypes.ChannelID]bool,
+	observationDueNanoseconds map[llotypes.ChannelID]uint64,
 	carryForward map[llotypes.StreamID]map[llotypes.Aggregator]*protocol.TimestampedStreamValue,
 ) error {
 	pb := &protocol.LLOHotStateProto{
@@ -307,6 +347,17 @@ func writeHotState(
 		return pb.ReportableChannelIDs[i] < pb.ReportableChannelIDs[j]
 	})
 
+	pb.ObservationDueNanoseconds = make([]*protocol.LLOChannelIDAndObservationDueProto, 0, len(observationDueNanoseconds))
+	for id, dueAt := range observationDueNanoseconds {
+		pb.ObservationDueNanoseconds = append(pb.ObservationDueNanoseconds, &protocol.LLOChannelIDAndObservationDueProto{
+			ChannelID:        id,
+			DueAtNanoseconds: dueAt,
+		})
+	}
+	sort.Slice(pb.ObservationDueNanoseconds, func(i, j int) bool {
+		return pb.ObservationDueNanoseconds[i].ChannelID < pb.ObservationDueNanoseconds[j].ChannelID
+	})
+
 	for sid, aggregates := range carryForward {
 		for agg, tsv := range aggregates {
 			if tsv == nil {

llo/dev/v31/kv_test.go

@@ -58,7 +58,7 @@ func Test_ChannelCache_StaleSeqNrForcesReload(t *testing.T) {
 	kv := newMemKV()
 	defs := llotypes.ChannelDefinitions{1: jsonChannel()}
 	require.NoError(t, writeChannelState(kv, 5, defs))
-	require.NoError(t, writeHotState(kv, 0, nil, nil, nil))
+	require.NoError(t, writeHotState(kv, 0, nil, nil, nil, nil))
 
 	cache := protocol.NewChannelCache()
 	s, err := loadKVState(kv, cache)
@@ -145,7 +145,7 @@ func Test_KVRecords_DeterministicAndRoundTrip(t *testing.T) {
 	for i := 0; i < 8; i++ {
 		kv := newMemKV()
 		require.NoError(t, writeChannelState(kv, 9, defs))
-		require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, carry))
+		require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, nil, carry))
 		if i == 0 {
 			channelBytes, hotBytes = kv.m[string(keyChannelState)], kv.m[string(keyHotState)]
 			continue
@@ -156,7 +156,7 @@ func Test_KVRecords_DeterministicAndRoundTrip(t *testing.T) {
 
 	kv := newMemKV()
 	require.NoError(t, writeChannelState(kv, 9, defs))
-	require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, carry))
+	require.NoError(t, writeHotState(kv, 1_234, validAfter, reportable, nil, carry))
 	require.Equal(t, uint64(9), binary.BigEndian.Uint64(kv.m[string(keyChannelSeqNr)]))
 
 	s, err := loadKVState(kv, nil)