Added include invalid pending queue items for the vault

pkg/capabilities/actions/vault/messages.proto

@@ -130,6 +130,12 @@ enum RequestType {
   LIST_SECRET_IDENTIFIERS = 5;
 }
 
+// ObservationError is an item-level Err(E) contribution for a pending queue request.
+// When set on Observation, the observation is an error contribution for that request id.
+message ObservationError {
+  string message = 1;
+}
+
 message Observation {
   string id = 1;
   RequestType request_type = 2;
@@ -149,12 +155,20 @@ message Observation {
     DeleteSecretsResponse delete_secrets_response = 11;
     ListSecretIdentifiersResponse list_secret_identifiers_response = 12;
   }
+
+  ObservationError error = 13;
+}
+
+enum PendingQueueStallSignal {
+  PENDING_QUEUE_STALL_SIGNAL_CONTINUE = 0;
+  PENDING_QUEUE_STALL_SIGNAL_STALLED = 1;
 }
 
 message Observations {
   repeated Observation observations = 1;
   repeated bytes pending_queue_items = 2;
   bytes sortNonce = 3;
+  PendingQueueStallSignal pending_queue_stall_signal = 4;
 }
 
 message Outcome {

pkg/capabilities/actions/vault/messages_test.go

@@ -0,0 +1,84 @@
+package vault
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestObservationErrorField_SHAStableForOkObservations(t *testing.T) {
+	t.Parallel()
+
+	obs := &Observation{
+		Id:          "request-1",
+		RequestType: RequestType_DELETE_SECRETS,
+		Response: &Observation_DeleteSecretsResponse{
+			DeleteSecretsResponse: &DeleteSecretsResponse{
+				Responses: []*DeleteSecretResponse{{Success: false}},
+			},
+		},
+	}
+
+	before, err := proto.MarshalOptions{Deterministic: true}.Marshal(obs)
+	require.NoError(t, err)
+
+	roundTrip := &Observation{}
+	require.NoError(t, proto.Unmarshal(before, roundTrip))
+	require.Nil(t, roundTrip.GetError())
+
+	after, err := proto.MarshalOptions{Deterministic: true}.Marshal(roundTrip)
+	require.NoError(t, err)
+	require.Equal(t, before, after)
+}
+
+func TestObservationErrorField_RoundTrip(t *testing.T) {
+	t.Parallel()
+
+	obs := &Observation{
+		Id:          "request-1",
+		RequestType: RequestType_CREATE_SECRETS,
+		Error: &ObservationError{
+			Message: "request is not valid",
+		},
+	}
+
+	b, err := proto.MarshalOptions{Deterministic: true}.Marshal(obs)
+	require.NoError(t, err)
+
+	out := &Observation{}
+	require.NoError(t, proto.Unmarshal(b, out))
+	require.Equal(t, "request is not valid", out.GetError().GetMessage())
+	require.Nil(t, out.GetCreateSecretsResponse())
+}
+
+func TestPendingQueueStallSignal_DefaultIsWireStable(t *testing.T) {
+	t.Parallel()
+
+	obs := &Observations{SortNonce: make([]byte, 32)}
+	before, err := proto.MarshalOptions{Deterministic: true}.Marshal(obs)
+	require.NoError(t, err)
+
+	roundTrip := &Observations{}
+	require.NoError(t, proto.Unmarshal(before, roundTrip))
+	require.Equal(t, PendingQueueStallSignal_PENDING_QUEUE_STALL_SIGNAL_CONTINUE, roundTrip.GetPendingQueueStallSignal())
+
+	after, err := proto.MarshalOptions{Deterministic: true}.Marshal(roundTrip)
+	require.NoError(t, err)
+	require.Equal(t, before, after)
+}
+
+func TestPendingQueueStallSignal_RoundTrip(t *testing.T) {
+	t.Parallel()
+
+	obs := &Observations{
+		SortNonce:               make([]byte, 32),
+		PendingQueueStallSignal: PendingQueueStallSignal_PENDING_QUEUE_STALL_SIGNAL_STALLED,
+	}
+	b, err := proto.MarshalOptions{Deterministic: true}.Marshal(obs)
+	require.NoError(t, err)
+
+	out := &Observations{}
+	require.NoError(t, proto.Unmarshal(b, out))
+	require.Equal(t, PendingQueueStallSignal_PENDING_QUEUE_STALL_SIGNAL_STALLED, out.GetPendingQueueStallSignal())
+}

pkg/settings/cresettings/README.md

@@ -49,6 +49,9 @@ flowchart
         VaultGetSecretsShareAggregationIncludesPublicKeys[/VaultGetSecretsShareAggregationIncludesPublicKeys\]:::gate
         VaultOwnerAddressCanonicalizationEnabled[/VaultOwnerAddressCanonicalizationEnabled\]:::gate
         VaultJSONOmitUnpopulatedEnabled[/VaultJSONOmitUnpopulatedEnabled\]:::gate
+        VaultGetSecretsRelaxedConsensusEnabled[/VaultGetSecretsRelaxedConsensusEnabled\]:::gate
+        VaultIncludeInvalidPendingItemsEnabled[/VaultIncludeInvalidPendingItemsEnabled\]:::gate
+        VaultPendingQueueStallThreshold{{VaultPendingQueueStallThreshold}}:::bound
         VaultSignedResponseRequestIDEnabled[/VaultSignedResponseRequestIDEnabled\]:::gate
     end
 

pkg/settings/cresettings/defaults.json

@@ -14,6 +14,9 @@
 	"VaultGetSecretsShareAggregationIncludesPublicKeys": "false",
 	"VaultOwnerAddressCanonicalizationEnabled": "false",
 	"VaultJSONOmitUnpopulatedEnabled": "false",
+	"VaultGetSecretsRelaxedConsensusEnabled": "false",
+	"VaultIncludeInvalidPendingItemsEnabled": "false",
+	"VaultPendingQueueStallThreshold": "0",
 	"VaultSignedResponseRequestIDEnabled": "false",
 	"GatewayHTTPGlobalRate": "500rps:500",
 	"GatewayHTTPPerNodeRate": "100rps:100",

pkg/settings/cresettings/defaults.toml

@@ -13,6 +13,9 @@ VaultOptimizationsEnabled = 'false'
 VaultGetSecretsShareAggregationIncludesPublicKeys = 'false'
 VaultOwnerAddressCanonicalizationEnabled = 'false'
 VaultJSONOmitUnpopulatedEnabled = 'false'
+VaultGetSecretsRelaxedConsensusEnabled = 'false'
+VaultIncludeInvalidPendingItemsEnabled = 'false'
+VaultPendingQueueStallThreshold = '0'
 VaultSignedResponseRequestIDEnabled = 'false'
 GatewayHTTPGlobalRate = '500rps:500'
 GatewayHTTPPerNodeRate = '100rps:100'

pkg/settings/cresettings/settings.go

@@ -68,6 +68,9 @@ var Default = Schema{
 	VaultGetSecretsShareAggregationIncludesPublicKeys: Bool(false),
 	VaultOwnerAddressCanonicalizationEnabled:          Bool(false),
 	VaultJSONOmitUnpopulatedEnabled:                   Bool(false),
+	VaultGetSecretsRelaxedConsensusEnabled:            Bool(false),
+	VaultIncludeInvalidPendingItemsEnabled:            Bool(false),
+	VaultPendingQueueStallThreshold:                   Int(0),
 	VaultSignedResponseRequestIDEnabled:               Bool(false),
 	GatewayHTTPGlobalRate:                             Rate(rate.Limit(500), 500),
 	GatewayHTTPPerNodeRate:                            Rate(rate.Limit(100), 100),
@@ -327,6 +330,9 @@ type Schema struct {
 	VaultGetSecretsShareAggregationIncludesPublicKeys Setting[bool]
 	VaultOwnerAddressCanonicalizationEnabled          Setting[bool]
 	VaultJSONOmitUnpopulatedEnabled                   Setting[bool]
+	VaultGetSecretsRelaxedConsensusEnabled            Setting[bool]
+	VaultIncludeInvalidPendingItemsEnabled            Setting[bool]
+	VaultPendingQueueStallThreshold                   Setting[int] `unit:"{observation}"`
 	VaultSignedResponseRequestIDEnabled               Setting[bool]
 	GatewayHTTPGlobalRate                             Setting[config.Rate]
 	GatewayHTTPPerNodeRate                            Setting[config.Rate]

pkg/settings/cresettings/settings_test.go

@@ -138,6 +138,9 @@ func TestSchema_Unmarshal(t *testing.T) {
 	assert.False(t, cfg.VaultGetSecretsShareAggregationIncludesPublicKeys.DefaultValue)
 	assert.False(t, cfg.VaultOwnerAddressCanonicalizationEnabled.DefaultValue)
 	assert.False(t, cfg.VaultJSONOmitUnpopulatedEnabled.DefaultValue)
+	assert.False(t, cfg.VaultGetSecretsRelaxedConsensusEnabled.DefaultValue)
+	assert.False(t, cfg.VaultIncludeInvalidPendingItemsEnabled.DefaultValue)
+	assert.Equal(t, 0, cfg.VaultPendingQueueStallThreshold.DefaultValue)
 	assert.False(t, cfg.VaultSignedResponseRequestIDEnabled.DefaultValue)
 	assert.Equal(t, config.Rate{Limit: rate.Limit(20), Burst: 7}, cfg.GatewayConfidentialRelayGlobalRate.DefaultValue)
 	assert.Equal(t, config.Rate{Limit: rate.Limit(4), Burst: 2}, cfg.GatewayConfidentialRelayPerNodeRate.DefaultValue)