[DONTime] Rollout flag for pruning fix

Author: bolekk

pkg/workflows/dontime/pb/dontime.pb.go

@@ -5,6 +5,9 @@ option go_package = "github.com/smartcontractkit/chainlink-common/pkg/workflows/
 message Observation {
   int64 timestamp = 1;
   map<string, int64> requests = 2;
+  // Flag to roll out execution pruning fix.
+  // TODO(CRE-2497): Remove after rollout.
+  bool prune_executions = 3;
 }
 
 message Observations {

pkg/workflows/dontime/pb/dontime.proto

@@ -7,6 +7,7 @@ import (
 	"slices"
 	"time"
 
+	"go.opentelemetry.io/otel/metric"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/structpb"
 
@@ -15,10 +16,51 @@ import (
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 	"github.com/smartcontractkit/libocr/quorumhelper"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb"
 )
 
+type pluginMetrics struct {
+	donTime        metric.Int64Gauge
+	donTimeEntries metric.Int64Gauge
+	outcomeSize    metric.Int64Gauge
+}
+
+func newPluginMetrics() (pluginMetrics, error) {
+	meter := beholder.GetMeter()
+
+	donTime, err := meter.Int64Gauge("dontime_outcome_don_time_ms",
+		metric.WithDescription("DON consensus timestamp included in the latest outcome, in milliseconds"),
+		metric.WithUnit("ms"),
+	)
+	if err != nil {
+		return pluginMetrics{}, fmt.Errorf("failed to create don_time gauge: %w", err)
+	}
+
+	donTimeEntries, err := meter.Int64Gauge("dontime_outcome_entries",
+		metric.WithDescription("Number of workflow execution entries tracked in the latest outcome"),
+		metric.WithUnit("{entry}"),
+	)
+	if err != nil {
+		return pluginMetrics{}, fmt.Errorf("failed to create don_time_entries gauge: %w", err)
+	}
+
+	outcomeSize, err := meter.Int64Gauge("dontime_outcome_size_bytes",
+		metric.WithDescription("Serialised size of the latest outcome in bytes"),
+		metric.WithUnit("By"),
+	)
+	if err != nil {
+		return pluginMetrics{}, fmt.Errorf("failed to create outcome_size gauge: %w", err)
+	}
+
+	return pluginMetrics{
+		donTime:        donTime,
+		donTimeEntries: donTimeEntries,
+		outcomeSize:    outcomeSize,
+	}, nil
+}
+
 type Plugin struct {
 	store          *Store
 	config         ocr3types.ReportingPluginConfig
@@ -27,6 +69,8 @@ type Plugin struct {
 
 	batchSize       int
 	minTimeIncrease int64
+
+	metrics pluginMetrics
 }
 
 var _ ocr3types.ReportingPlugin[[]byte] = (*Plugin)(nil)
@@ -42,13 +86,19 @@ func NewPlugin(store *Store, config ocr3types.ReportingPluginConfig, offchainCfg
 		return nil, errors.New("execution removal time must be positive")
 	}
 
+	metrics, err := newPluginMetrics()
+	if err != nil {
+		return nil, err
+	}
+
 	return &Plugin{
 		store:           store,
 		config:          config,
 		offChainConfig:  offchainCfg,
 		lggr:            logger.Named(lggr, "DONTimePlugin"),
 		batchSize:       int(offchainCfg.MaxBatchSize),
 		minTimeIncrease: offchainCfg.MinTimeIncrease / int64(time.Millisecond),
+		metrics:         metrics,
 	}, nil
 }
 
@@ -97,8 +147,9 @@ func (p *Plugin) Observation(_ context.Context, outctx ocr3types.OutcomeContext,
 	}
 
 	observation := &pb.Observation{
-		Timestamp: time.Now().UTC().UnixMilli(),
-		Requests:  requests,
+		Timestamp:       time.Now().UTC().UnixMilli(),
+		Requests:        requests,
+		PruneExecutions: true,
 	}
 
 	return proto.MarshalOptions{Deterministic: true}.Marshal(observation)
@@ -112,7 +163,7 @@ func (p *Plugin) ObservationQuorum(_ context.Context, _ ocr3types.OutcomeContext
 	return quorumhelper.ObservationCountReachesObservationQuorum(quorumhelper.QuorumTwoFPlusOne, p.config.N, p.config.F, aos), nil
 }
 
-func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ types.Query, aos []types.AttributedObservation) (ocr3types.Outcome, error) {
+func (p *Plugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, _ types.Query, aos []types.AttributedObservation) (ocr3types.Outcome, error) {
 	observationCounts := map[string]int64{} // counts how many nodes reported where a new DON timestamp might be needed
 	type timestampNodePair struct {
 		Timestamp        int64
@@ -129,14 +180,33 @@ func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ t
 		prevOutcome.ObservedDonTimes = make(map[string]*pb.ObservedDonTimes)
 	}
 
+	// Unmarshal all observations once and compute pruneExecutions.
+	// Only prune when all nodes are updated. Even if this rolls back, the logic is still correct.
+	parsedAOs := make([]*pb.Observation, len(aos))
+	pruneExecutions := true
 	for idx, ao := range aos {
 		observation := &pb.Observation{}
 		if err := proto.Unmarshal(ao.Observation, observation); err != nil {
 			p.lggr.Errorf("failed to unmarshal observation in Outcome phase")
 			continue
 		}
+		parsedAOs[idx] = observation
+		if !observation.PruneExecutions {
+			pruneExecutions = false // need all nodes to agree
+		}
+	}
+
+	for idx, observation := range parsedAOs {
+		if observation == nil {
+			continue
+		}
 
 		for id, requestSeqNum := range observation.Requests {
+			if !pruneExecutions { // TODO(CRE-2497): legacy behavior, remove after rollout
+				if _, ok := prevOutcome.ObservedDonTimes[id]; !ok {
+					prevOutcome.ObservedDonTimes[id] = &pb.ObservedDonTimes{}
+				}
+			}
 			var currSeqNum int64
 			if times, ok := prevOutcome.ObservedDonTimes[id]; ok {
 				currSeqNum = int64(len(times.Timestamps))
@@ -196,14 +266,23 @@ func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ t
 
 	// Remove expired and empty workflow executions
 	for id, observedTimes := range outcome.ObservedDonTimes {
-		if observedTimes == nil || len(observedTimes.Timestamps) == 0 {
-			delete(outcome.ObservedDonTimes, id)
-			p.store.deleteExecutionID(id)
-			continue
-		}
-		if donTime >= observedTimes.Timestamps[0]+p.offChainConfig.ExecutionRemovalTime.AsDuration().Milliseconds() {
-			delete(outcome.ObservedDonTimes, id)
-			p.store.deleteExecutionID(id)
+		if !pruneExecutions { // TODO(CRE-2497): legacy behavior, remove after rollout
+			if observedTimes != nil && len(observedTimes.Timestamps) > 0 {
+				if donTime >= observedTimes.Timestamps[0]+p.offChainConfig.ExecutionRemovalTime.AsDuration().Milliseconds() {
+					delete(outcome.ObservedDonTimes, id)
+					p.store.deleteExecutionID(id)
+				}
+			}
+		} else {
+			if observedTimes == nil || len(observedTimes.Timestamps) == 0 {
+				delete(outcome.ObservedDonTimes, id)
+				p.store.deleteExecutionID(id)
+				continue
+			}
+			if donTime >= observedTimes.Timestamps[0]+p.offChainConfig.ExecutionRemovalTime.AsDuration().Milliseconds() {
+				delete(outcome.ObservedDonTimes, id)
+				p.store.deleteExecutionID(id)
+			}
 		}
 	}
 
@@ -212,6 +291,9 @@ func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ t
 		"observedDonTimesEntries", len(outcome.ObservedDonTimes),
 		"outcomeSizeBytes", len(outcomeBytes),
 	)
+	p.metrics.donTime.Record(ctx, outcome.Timestamp)
+	p.metrics.donTimeEntries.Record(ctx, int64(len(outcome.ObservedDonTimes)))
+	p.metrics.outcomeSize.Record(ctx, int64(len(outcomeBytes)))
 	return outcomeBytes, err
 }
 

pkg/workflows/dontime/plugin.go

@@ -159,28 +159,24 @@ func TestPlugin_Outcome(t *testing.T) {
 	timestamp := time.Now().UnixMilli()
 	observations := []*pb.Observation{
 		{
-			Timestamp: timestamp,
-			Requests: map[string]int64{
-				executionID: 0,
-			},
+			Timestamp:       timestamp,
+			Requests:        map[string]int64{executionID: 0},
+			PruneExecutions: true,
 		},
 		{
-			Timestamp: timestamp - int64(time.Second),
-			Requests: map[string]int64{
-				executionID: 0,
-			},
+			Timestamp:       timestamp - int64(time.Second),
+			Requests:        map[string]int64{executionID: 0},
+			PruneExecutions: true,
 		},
 		{
-			Timestamp: timestamp + int64(time.Second),
-			Requests: map[string]int64{
-				executionID: 0,
-			},
+			Timestamp:       timestamp + int64(time.Second),
+			Requests:        map[string]int64{executionID: 0},
+			PruneExecutions: true,
 		},
 		{
-			Timestamp: timestamp,
-			Requests: map[string]int64{
-				executionID: 0,
-			},
+			Timestamp:       timestamp,
+			Requests:        map[string]int64{executionID: 0},
+			PruneExecutions: true,
 		},
 	}
 
@@ -224,8 +220,9 @@ func TestPlugin_Outcome_SequenceNumberHandling(t *testing.T) {
 		aos := make([]types.AttributedObservation, numNodes)
 		for i := 0; i < numNodes; i++ {
 			obs := &pb.Observation{
-				Timestamp: timestamp + int64(i),
-				Requests:  requests,
+				Timestamp:       timestamp + int64(i),
+				Requests:        requests,
+				PruneExecutions: true,
 			}
 			rawObs, err := proto.Marshal(obs)
 			require.NoError(t, err)
@@ -434,22 +431,10 @@ func TestPlugin_FinishedExecutions(t *testing.T) {
 	t.Run("Outcome: remove expired workflow executions", func(t *testing.T) {
 		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{},
-			},
+			{Timestamp: timestamp, Requests: map[string]int64{}, PruneExecutions: true},
+			{Timestamp: timestamp - int64(time.Second), Requests: map[string]int64{}, PruneExecutions: true},
+			{Timestamp: timestamp + int64(time.Second), Requests: map[string]int64{}, PruneExecutions: true},
+			{Timestamp: timestamp, Requests: map[string]int64{}, PruneExecutions: true},
 		}
 
 		aos := make([]types.AttributedObservation, 4)
@@ -484,6 +469,47 @@ func TestPlugin_FinishedExecutions(t *testing.T) {
 		require.NotContains(t, outcomeProto.ObservedDonTimes, "workflow-123")
 	})
 
+	t.Run("Outcome: legacy path when only half nodes have PruneExecutions set", func(t *testing.T) {
+		timestamp := time.Now().UnixMilli()
+		emptyID := "empty-workflow"
+
+		// Only 2 of 4 nodes have PruneExecutions=true → pruneExecutions stays false → legacy path.
+		observations := []*pb.Observation{
+			{Timestamp: timestamp, Requests: map[string]int64{}, PruneExecutions: true},
+			{Timestamp: timestamp - int64(time.Second), Requests: map[string]int64{}, PruneExecutions: true},
+			{Timestamp: timestamp + int64(time.Second), Requests: map[string]int64{}, PruneExecutions: false},
+			{Timestamp: timestamp, Requests: map[string]int64{}, PruneExecutions: false},
+		}
+
+		aos := make([]types.AttributedObservation, len(observations))
+		for i, obs := range observations {
+			rawObs, err := proto.Marshal(obs)
+			require.NoError(t, err)
+			aos[i] = types.AttributedObservation{Observation: rawObs, Observer: commontypes.OracleID(i)}
+		}
+
+		// prevOutcome contains an entry for emptyID with no timestamps.
+		prevOutcome := &pb.Outcome{
+			Timestamp: timestamp - 1000,
+			ObservedDonTimes: map[string]*pb.ObservedDonTimes{
+				emptyID: {Timestamps: []int64{}},
+			},
+		}
+		prevOutcomeBytes, err := proto.Marshal(prevOutcome)
+		require.NoError(t, err)
+
+		outcome, err := plugin.Outcome(ctx, ocr3types.OutcomeContext{PreviousOutcome: prevOutcomeBytes}, query, aos)
+		require.NoError(t, err)
+
+		legacyOutcomeProto := &pb.Outcome{}
+		err = proto.Unmarshal(outcome, legacyOutcomeProto)
+		require.NoError(t, err)
+
+		// Legacy behavior: empty-timestamps entry is NOT pruned.
+		require.Contains(t, legacyOutcomeProto.ObservedDonTimes, emptyID)
+		require.Empty(t, legacyOutcomeProto.ObservedDonTimes[emptyID].Timestamps)
+	})
+
 	t.Run("Transmit: delete removed executionIDs", func(t *testing.T) {
 		store.setDonTimes("workflow-123", []int64{time.Now().UnixMilli()})