Changes required for consensus capability duplicate outcomes protection (#1622)

* changes to prevent consensus capability duplicate outcome

Author: ettec

pkg/capabilities/consensus/ocr3/types/ocr3_config_types.pb.go

@@ -18,4 +18,5 @@ message ReportingPluginConfig {
   uint32 maxBatchSize = 6;
   uint64 outcomePruningThreshold = 7;
   google.protobuf.Duration requestTimeout = 8;
+  uint64 historical_outcome_expiry_seq_nr_span = 9;
 }

pkg/capabilities/consensus/ocr3/types/ocr3_config_types.proto

@@ -68,6 +68,11 @@ func (s *Store[T]) GetByIDs(requestIDs []string) []T {
 func (s *Store[T]) FirstN(batchSize int) ([]T, error) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
+
+	return s.firstN(batchSize)
+}
+
+func (s *Store[T]) firstN(batchSize int) ([]T, error) {
 	if batchSize == 0 {
 		return nil, errors.New("batchsize cannot be 0")
 	}
@@ -91,6 +96,19 @@ func (s *Store[T]) FirstN(batchSize int) ([]T, error) {
 	return got, nil
 }
 
+// All retrieves all requests.
+// The method deep-copies requests before returning them.
+func (s *Store[T]) All() ([]T, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	allRequestsCnt := len(s.requestIDs)
+	if allRequestsCnt == 0 {
+		return []T{}, nil
+	}
+
+	return s.firstN(allRequestsCnt)
+}
+
 // RangeN retrieves up to `batchSize` requests starting at index `start`.
 // It deep-copies each request before returning.
 func (s *Store[T]) RangeN(start, batchSize int) ([]T, error) {

pkg/capabilities/consensus/requests/store.go

@@ -25,6 +25,12 @@ func TestOCR3Store(t *testing.T) {
 		ExpiresAt:           n.Add(10 * time.Second),
 	}
 
+	t.Run("all with 0 requests", func(t *testing.T) {
+		items, err := s.All()
+		require.NoError(t, err)
+		assert.Len(t, items, 0)
+	})
+
 	t.Run("Add", func(t *testing.T) {
 		err := s.Add(req)
 		require.NoError(t, err)
@@ -64,6 +70,12 @@ func TestOCR3Store(t *testing.T) {
 		assert.Len(t, items, 10)
 	})
 
+	t.Run("all with 10 requests", func(t *testing.T) {
+		items, err := s.All()
+		require.NoError(t, err)
+		assert.Len(t, items, 10)
+	})
+
 	t.Run("rangeN", func(t *testing.T) {
 		err := s.Add(&ocr3.ReportRequest{WorkflowExecutionID: uuid.New().String(), ExpiresAt: n.Add(1 * time.Hour)})
 		r, err := s.RangeN(0, 1)