llo/calculated: ensure stream dependencies and aggregates are present

Author: brunotm

llo/plugin_reports.go

@@ -55,7 +55,11 @@ func (p *Plugin) reports(ctx context.Context, seqNr uint64, rawOutcome ocr3types
 		cd := outcome.ChannelDefinitions[cid]
 		values := make([]StreamValue, 0, len(cd.Streams))
 		for _, strm := range cd.Streams {
-			values = append(values, outcome.StreamAggregates[strm.StreamID][strm.Aggregator])
+			if aggregates, ok := outcome.StreamAggregates[strm.StreamID]; !ok {
+				values = append(values, nil)
+			} else {
+				values = append(values, aggregates[strm.Aggregator])
+			}
 		}
 
 		report := Report{

llo/reportcodecs/evm/report_codec_evm_abi_encode_unpacked_expr.go

@@ -46,7 +46,16 @@ func (r ReportCodecEVMABIEncodeUnpackedExpr) Encode(report llo.Report, cd llotyp
 	// as Opts is small.
 	opts := ReportFormatEVMABIEncodeOpts{}
 	if err = (&opts).Decode(cd.Opts); err != nil {
-		return nil, fmt.Errorf("failed to decode opts; got: '%s'; %w", cd.Opts, err)
+		return nil, fmt.Errorf("ReportCodecEVMABIEncodeUnpackedExpr failed to decode opts; got: '%s'; %w", cd.Opts, err)
+	}
+
+	if len(opts.ABI) < 1 {
+		return nil, fmt.Errorf("ReportCodecEVMABIEncodeUnpackedExpr no expressions found in channel definition")
+	}
+
+	// not enough streams for calculated feed
+	if cd.Streams[len(cd.Streams)-1].StreamID != opts.ABI[len(opts.ABI)-1].encoders[0].ExpressionStreamID {
+		return nil, fmt.Errorf("ReportCodecEVMABIEncodeUnpackedExpr not enough streams for calculated streams; expected: %d, got: %d", opts.ABI[len(opts.ABI)-1].encoders[0].ExpressionStreamID, len(cd.Streams))
 	}
 
 	validAfter := ConvertTimestamp(report.ValidAfterNanoseconds, opts.TimestampPrecision)

llo/reportcodecs/evm/report_codec_evm_abi_encode_unpacked_expr_test.go

@@ -74,7 +74,7 @@ func TestReportCodecEVMABIEncodeUnpackedExpr_Encode(t *testing.T) {
 		codec := ReportCodecEVMABIEncodeUnpackedExpr{}
 		_, err = codec.Encode(report, cd)
 		require.Error(t, err)
-		assert.Contains(t, err.Error(), "ABI and values length mismatch")
+		assert.Contains(t, err.Error(), "ReportCodecEVMABIEncodeUnpackedExpr no expressions found in channel definition")
 	})
 
 	t.Run("DEX-based asset schema example", func(t *testing.T) {

llo/stream_calculated.go

@@ -585,6 +585,11 @@ func (p *Plugin) ProcessCalculatedStreams(outcome *Outcome) {
 
 		}
 
+		if len(cd.Streams) < 1 {
+			p.Logger.Errorw("no streams or expressions found in channel definition", "channelID", cid)
+			continue
+		}
+
 		// channel definitions are inherited from the previous outcome,
 		// so we only update the channel definition streams if we haven't done it before
 		if cd.Streams[len(cd.Streams)-1].StreamID != copt.ABI[len(copt.ABI)-1].ExpressionStreamID {