llo: calculated streams should skip observations and aggregations (#183)

Author: brunotm

llo/plugin_observation.go

@@ -134,6 +134,10 @@ func (p *Plugin) observation(ctx context.Context, outctx ocr3types.OutcomeContex
 			obs.StreamValues = make(StreamValues)
 			for _, channelDefinition := range previousOutcome.ChannelDefinitions {
 				for _, strm := range channelDefinition.Streams {
+					// Calculated streams have no values to observe
+					if strm.Aggregator == llotypes.AggregatorCalculated {
+						continue
+					}
 					obs.StreamValues[strm.StreamID] = nil
 				}
 			}

llo/plugin_outcome.go

@@ -204,6 +204,12 @@ func (p *Plugin) outcome(outctx ocr3types.OutcomeContext, query types.Query, aos
 	for cid, cd := range outcome.ChannelDefinitions {
 		for _, strm := range cd.Streams {
 			sid, agg := strm.StreamID, strm.Aggregator
+
+			// Calculated streams are handled after all streams are aggregated
+			if strm.Aggregator == llotypes.AggregatorCalculated {
+				continue
+			}
+
 			if _, exists := outcome.StreamAggregates[sid][agg]; exists {
 				// Should only happen in the case of duplicate
 				// streams, no need to aggregate twice.

llo/stream_calculated.go

@@ -436,6 +436,10 @@ func (p *Plugin) ProcessCalculatedStreams(outcome *Outcome) {
 		var err error
 		env := NewEnv(outcome)
 		for _, stream := range cd.Streams {
+			if stream.Aggregator == llotypes.AggregatorCalculated {
+				continue
+			}
+
 			p.Logger.Debugw("setting stream value", "channelID", cid, "streamID", stream.StreamID, "aggregator", stream.Aggregator)
 
 			if err = env.SetStreamValue(stream.StreamID, outcome.StreamAggregates[stream.StreamID][stream.Aggregator]); err != nil {