test: cover dontime outcome changes

Author: cawthorne

pkg/loop/mocks/relayer.go

@@ -41,6 +41,7 @@ type Relayer interface {
 	NewContractReader(_ context.Context, contractReaderConfig []byte) (types.ContractReader, error)
 	NewContractWriter(_ context.Context, contractWriterConfig []byte) (types.ContractWriter, error)
 	LatestHead(context.Context) (types.Head, error)
+	FinalizedHead(context.Context) (types.Head, error)
 	GetChainInfo(ctx context.Context) (types.ChainInfo, error)
 }
 

pkg/types/core/mocks/relayer.go

@@ -101,6 +101,9 @@ type ChainService interface {
 
 	// LatestHead returns the latest head for the underlying chain.
 	LatestHead(ctx context.Context) (Head, error)
+	// FinalizedHead returns the finalized head for the underlying chain. Implementations
+	// that do not distinguish finalized heads may return the latest observed head.
+	FinalizedHead(ctx context.Context) (Head, error)
 	// GetChainInfo returns the ChainInfo for this Relayer.
 	GetChainInfo(ctx context.Context) (ChainInfo, error)
 	// GetChainStatus returns the ChainStatus for this Relayer.
@@ -115,8 +118,8 @@ type ChainService interface {
 }
 
 // FinalizedHeadService is implemented by relayers that can distinguish finalized
-// heads from the latest observed head. Callers should fall back to LatestHead
-// when a chain does not implement this interface.
+// heads from the latest observed head. This remains useful for callers operating
+// on generic service values rather than the full ChainService interface.
 type FinalizedHeadService interface {
 	FinalizedHead(ctx context.Context) (Head, error)
 }

pkg/types/core/relayerset.go

@@ -213,6 +213,101 @@ func TestPlugin_Outcome(t *testing.T) {
 	require.Equal(t, []int64{timestamp}, outcomeProto.ObservedDonTimes[executionID].Timestamps)
 }
 
+func TestPlugin_OutcomeInitializesMissingObservedDonTimesEntry(t *testing.T) {
+	lggr := logger.Test(t)
+	store := NewStore(DefaultRequestTimeout)
+	config, offchainCfg := newTestPluginConfig(t), newTestPluginOffchainConfig(t)
+	ctx := t.Context()
+
+	plugin, err := NewPlugin(store, config, offchainCfg, lggr)
+	require.NoError(t, err)
+
+	query, err := plugin.Query(ctx, ocr3types.OutcomeContext{PreviousOutcome: []byte("")})
+	require.NoError(t, err)
+
+	const executionID = "workflow-missing-prev-entry"
+	timestamp := time.Now().UnixMilli()
+	observations := []*pb.Observation{
+		{Timestamp: timestamp, Requests: map[string]int64{executionID: 0}},
+		{Timestamp: timestamp - int64(time.Second), Requests: map[string]int64{executionID: 0}},
+		{Timestamp: timestamp + int64(time.Second), Requests: map[string]int64{executionID: 0}},
+		{Timestamp: timestamp, Requests: map[string]int64{executionID: 0}},
+	}
+
+	aos := make([]types.AttributedObservation, len(observations))
+	for i, observation := range observations {
+		rawObs, marshalErr := proto.Marshal(observation)
+		require.NoError(t, marshalErr)
+		aos[i] = types.AttributedObservation{
+			Observation: rawObs,
+			Observer:    commontypes.OracleID(1),
+		}
+	}
+
+	prevOutcome := &pb.Outcome{
+		Timestamp:        0,
+		ObservedDonTimes: map[string]*pb.ObservedDonTimes{},
+	}
+	prevOutcomeBytes, err := proto.Marshal(prevOutcome)
+	require.NoError(t, err)
+
+	outcome, err := plugin.Outcome(ctx, ocr3types.OutcomeContext{PreviousOutcome: prevOutcomeBytes}, query, aos)
+	require.NoError(t, err)
+
+	outcomeProto := &pb.Outcome{}
+	require.NoError(t, proto.Unmarshal(outcome, outcomeProto))
+	require.Contains(t, outcomeProto.ObservedDonTimes, executionID)
+	require.Equal(t, []int64{timestamp}, outcomeProto.ObservedDonTimes[executionID].Timestamps)
+}
+
+func TestPlugin_OutcomeKeepsEmptyObservedDonTimesEntries(t *testing.T) {
+	lggr := logger.Test(t)
+	store := NewStore(DefaultRequestTimeout)
+	config, offchainCfg := newTestPluginConfig(t), newTestPluginOffchainConfig(t)
+	ctx := t.Context()
+
+	plugin, err := NewPlugin(store, config, offchainCfg, lggr)
+	require.NoError(t, err)
+
+	query, err := plugin.Query(ctx, ocr3types.OutcomeContext{PreviousOutcome: []byte("")})
+	require.NoError(t, err)
+
+	timestamp := time.Now().UnixMilli()
+	observations := []*pb.Observation{
+		{Timestamp: timestamp, Requests: map[string]int64{}},
+		{Timestamp: timestamp - int64(time.Second), Requests: map[string]int64{}},
+		{Timestamp: timestamp + int64(time.Second), Requests: map[string]int64{}},
+		{Timestamp: timestamp, Requests: map[string]int64{}},
+	}
+
+	aos := make([]types.AttributedObservation, len(observations))
+	for i, observation := range observations {
+		rawObs, marshalErr := proto.Marshal(observation)
+		require.NoError(t, marshalErr)
+		aos[i] = types.AttributedObservation{
+			Observation: rawObs,
+			Observer:    commontypes.OracleID(1),
+		}
+	}
+
+	prevOutcome := &pb.Outcome{
+		Timestamp: timestamp - int64(time.Second),
+		ObservedDonTimes: map[string]*pb.ObservedDonTimes{
+			"workflow-empty": {},
+		},
+	}
+	prevOutcomeBytes, err := proto.Marshal(prevOutcome)
+	require.NoError(t, err)
+
+	outcome, err := plugin.Outcome(ctx, ocr3types.OutcomeContext{PreviousOutcome: prevOutcomeBytes}, query, aos)
+	require.NoError(t, err)
+
+	outcomeProto := &pb.Outcome{}
+	require.NoError(t, proto.Unmarshal(outcome, outcomeProto))
+	require.Contains(t, outcomeProto.ObservedDonTimes, "workflow-empty")
+	require.Empty(t, outcomeProto.ObservedDonTimes["workflow-empty"].Timestamps)
+}
+
 func TestPlugin_FinishedExecutions(t *testing.T) {
 	lggr := logger.Test(t)
 	store := NewStore(DefaultRequestTimeout)

pkg/types/relayer.go

@@ -51,3 +51,37 @@ func TestTransmitter_TransmitDonTimeRequest(t *testing.T) {
 
 	require.Empty(t, store.GetRequest(executionID))
 }
+
+func TestTransmitter_TransmitPreservesCachedDonTimesForOmittedExecutionIDs(t *testing.T) {
+	lggr := logger.Test(t)
+	store := NewStore(DefaultRequestTimeout)
+	ctx := t.Context()
+
+	transmitter := NewTransmitter(lggr, store, "")
+
+	store.setDonTimes("workflow-stale", []int64{11, 22})
+
+	timestamp := time.Now().UnixMilli()
+	outcome := &pb.Outcome{
+		Timestamp: timestamp,
+		ObservedDonTimes: map[string]*pb.ObservedDonTimes{
+			"workflow-fresh": {Timestamps: []int64{timestamp}},
+		},
+	}
+
+	r := ocr3types.ReportWithInfo[[]byte]{}
+	var err error
+	r.Report, err = proto.Marshal(outcome)
+	require.NoError(t, err)
+
+	err = transmitter.Transmit(ctx, types.ConfigDigest{}, 0, r, []types.AttributedOnchainSignature{})
+	require.NoError(t, err)
+
+	staleDonTimes, err := store.GetDonTimes("workflow-stale")
+	require.NoError(t, err)
+	require.Equal(t, []int64{11, 22}, staleDonTimes)
+
+	freshDonTimes, err := store.GetDonTimes("workflow-fresh")
+	require.NoError(t, err)
+	require.Equal(t, []int64{timestamp}, freshDonTimes)
+}