test: add new E2E test to check pooling recovery

Author: adilsitos

e2e/proxy/proxy_test.go

@@ -26,6 +26,7 @@ type ProxyTestConfig struct {
 	TLSKeyFile                   string
 	AccessTokenCheckInterval     string
 	StaticSecretsRefreshInterval string
+	PollingFallbackInterval      string
 	UseSSE                       bool
 	ClientID                     string
 	ClientSecret                 string
@@ -45,6 +46,7 @@ func DefaultProxyTestConfig() ProxyTestConfig {
 func SSEProxyTestConfig() ProxyTestConfig {
 	config := DefaultProxyTestConfig()
 	config.UseSSE = true
+	config.PollingFallbackInterval = "10s"
 	return config
 }
 
@@ -69,6 +71,9 @@ func startProxy(t *testing.T, ctx context.Context, infisicalURL string, config P
 		args = append(args, "--event-subscription-enabled")
 		args = append(args, "--client-id", config.ClientID)
 		args = append(args, "--client-secret", config.ClientSecret)
+		if config.PollingFallbackInterval != "" {
+			args = append(args, "--polling-fallback-interval", config.PollingFallbackInterval)
+		}
 	}
 
 	proxyCmd := helpers.Command{
@@ -884,3 +889,91 @@ func TestProxy_SSEMultipleProjects(t *testing.T) {
 	}
 	require.True(t, foundOriginal, "Original secret not found in project 2")
 }
+func TestProxy_SSEPollingFallbackRecovery(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	infisical, helper, proxyCmd, _, _, identity := setupProxyTest(t, ctx, SSEProxyTestConfig())
+
+	result := helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
+		EnsureCmdRunning: proxyCmd,
+		ExpectedString:   "SSE manager initialized",
+		Timeout:          30 * time.Second,
+	})
+	require.Equal(t, helpers.WaitSuccess, result)
+
+	// create secret via API (not proxy)
+	initialSecret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
+		Prefix: "SSE_POLLING_FALLBACK_RECOVERY_",
+	})
+	helper.CreateSecretWithApi(ctx, initialSecret)
+
+	// fetch via proxy -> cache miss, now cached
+	slog.Info("Fetching secret via proxy (expecting cache miss)")
+	resp1 := helper.GetSecretsWithProxy(ctx)
+	require.Equal(t, http.StatusOK, resp1.StatusCode())
+	require.NotNil(t, resp1.JSON200)
+
+	var foundInitial bool
+	for _, s := range resp1.JSON200.Secrets {
+		if s.SecretKey == initialSecret.SecretKey {
+			assert.Equal(t, initialSecret.SecretValue, s.SecretValue)
+			foundInitial = true
+			break
+		}
+	}
+	require.True(t, foundInitial, "Initial secret not found in response")
+
+	// wait for SSE connection (demand-driven, triggered by first fetch)
+	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
+		EnsureCmdRunning: proxyCmd,
+		ExpectedString:   "SSE connection established",
+		Timeout:          30 * time.Second,
+		Interval:         200 * time.Millisecond,
+	})
+	require.Equal(t, helpers.WaitSuccess, result)
+
+	// change identity role to no-access to break SSE connection with auth errors
+	slog.Info("Changing identity role to no-access to break SSE connection")
+	infisical.UpdateMachineIdentityRole(t, ctx, identity.Id, "no-access")
+
+	// update secret via API so SSE event is triggered
+	// do it twice so we don't have race conditions to update the role
+	updatedSecret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
+		PresetName: initialSecret.SecretKey,
+	})
+	helper.UpdateSecretWithApi(ctx, updatedSecret)
+
+	updatedSecret2 := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
+		PresetName: updatedSecret.SecretKey,
+	})
+	helper.UpdateSecretWithApi(ctx, updatedSecret2)
+
+	// wait for SSE retries to exhaust and polling fallback to start
+	slog.Info("Waiting for SSE retries to exhaust and polling fallback to start")
+	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
+		EnsureCmdRunning: proxyCmd,
+		ExpectedString:   "SSE connection retries exhausted, transitioning to polling fallback",
+		Timeout:          20 * time.Second,
+		Interval:         200 * time.Millisecond,
+	})
+	require.Equal(t, helpers.WaitSuccess, result, "Should transition to polling fallback after retries exhausted")
+
+	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
+		EnsureCmdRunning: proxyCmd,
+		ExpectedString:   "Starting polling fallback for project",
+		Timeout:          10 * time.Second,
+		Interval:         200 * time.Millisecond,
+	})
+	require.Equal(t, helpers.WaitSuccess, result, "Polling fallback loop should start")
+
+	// restore identity role to member to allow SSE reconnection
+	slog.Info("Restoring identity role to member")
+	infisical.UpdateMachineIdentityRole(t, ctx, identity.Id, "member")
+
+	// verify proxy is still functional
+	slog.Info("Verifying proxy still works after SSE recovery from polling")
+	respFinal := helper.GetSecretsWithProxy(ctx)
+	require.Equal(t, http.StatusOK, respFinal.StatusCode())
+	require.True(t, proxyCmd.IsRunning(), "Proxy should still be running")
+}

infisical-cli

@@ -57,12 +57,12 @@ var proxyDebugCmd = &cobra.Command{
 	Hidden:                true,
 }
 
-func initializeSSEManager(ctx context.Context, clientId, clientSecret string, cache *proxy.Cache, domainURL *url.URL, streamingClient *http.Client, httpClient *http.Client) (*proxy.SSEManager, error) {
+func initializeSSEManager(ctx context.Context, clientId, clientSecret string, cache *proxy.Cache, domainURL *url.URL, streamingClient *http.Client, httpClient *http.Client, pollingFallbackInterval time.Duration) (*proxy.SSEManager, error) {
 	sseAuthState, err := proxy.NewSSEAuthState(clientId, clientSecret, domainURL, httpClient)
 	if err != nil {
 		return nil, err
 	}
-	return proxy.NewSSEManager(context.Background(), cache, domainURL, streamingClient, httpClient, sseAuthState), nil
+	return proxy.NewSSEManager(context.Background(), cache, domainURL, streamingClient, httpClient, sseAuthState, pollingFallbackInterval), nil
 }
 
 func startProxyServer(cmd *cobra.Command, args []string) {
@@ -132,6 +132,16 @@ func startProxyServer(cmd *cobra.Command, args []string) {
 		util.PrintErrorMessageAndExit(fmt.Sprintf("Invalid static-secrets-refresh-interval format '%s'. Use formats like 30m, 1h, 1d", staticSecretsRefreshIntervalStr))
 	}
 
+	pollingFallbackIntervalStr, err := cmd.Flags().GetString("polling-fallback-interval")
+	if err != nil {
+		util.HandleError(err, "Unable to parse polling-fallback-interval flag")
+	}
+
+	pollingFallbackInterval, err := util.ParseTimeDurationString(pollingFallbackIntervalStr, true)
+	if err != nil {
+		util.PrintErrorMessageAndExit(fmt.Sprintf("Invalid polling-fallback-interval format '%s'. Use formats like 1m, 5m, 10m", pollingFallbackIntervalStr))
+	}
+
 	useSSE, err := cmd.Flags().GetBool("event-subscription-enabled")
 	if err != nil {
 		util.HandleError(err, "Unable to parse event-subscription-enabled flag")
@@ -204,7 +214,7 @@ func startProxyServer(cmd *cobra.Command, args []string) {
 
 	var sseManager *proxy.SSEManager
 	if useSSE {
-		sseManager, err = initializeSSEManager(context.Background(), clientId, clientSecret, cache, domainURL, streamingClient, httpClient)
+		sseManager, err = initializeSSEManager(context.Background(), clientId, clientSecret, cache, domainURL, streamingClient, httpClient, pollingFallbackInterval)
 		if err != nil {
 			util.HandleError(err, "Failed to initialize SSE manager")
 		}
@@ -626,6 +636,7 @@ func init() {
 	proxyStartCmd.Flags().Bool("event-subscription-enabled", false, "Enable Event Subscription mode for real-time cache invalidation. When enabled, the static secrets refresh loop is disabled. If event subscriptions are unavailable, the proxy will fall back to a polling mechanism.  `--client id` and `--client-secret` are required when this is set to true ")
 	proxyStartCmd.Flags().String("client-id", "", "Machine identity universal auth client ID. This is required when using event subscriptions.")
 	proxyStartCmd.Flags().String("client-secret", "", "Machine identity universal auth client secret. This is required when using event subscriptions.")
+	proxyStartCmd.Flags().String("polling-fallback-interval", "10m", "How often to poll for secret changes when SSE is unavailable (e.g., 1m, 5m). Defaults to 5m. Only used when --event-subscription-enabled is set.")
 
 	proxyDebugCmd.Flags().String("listen-address", "localhost:8081", "The address where the proxy server is listening. Defaults to localhost:8081")
 

packages/cmd/proxy.go

@@ -362,6 +362,7 @@ func runProjectSecretsRefresh(cache *Cache, domainURL *url.URL, httpClient *http
 }
 
 func startProjectPollingLoop(ctx context.Context, cache *Cache, domainURL *url.URL, httpClient *http.Client, projectId, envSlug string, interval time.Duration, onPollComplete func()) {
+	log.Info().Str("projectId", projectId).Str("envSlug", envSlug).Msg("Starting project polling loop")
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 

packages/proxy/resync.go

@@ -188,28 +188,29 @@ type pollingState struct {
 // When SSE connections fail repeatedly for a project, the manager transitions
 // that project to a polling fallback and tracks it in pollingProjects.
 type SSEManager struct {
-	mu               sync.Mutex
-	connections      map[string]*SSEConnection // active SSE connections
-	pollingProjects  map[string]*pollingState  // projects in polling fallback
-	cache            *Cache
-	domainURL        *url.URL
-	httpClient       *http.Client // streaming client (no timeout) for SSE connections
-	resyncHttpClient *http.Client // regular client (with timeout) for cache resync requests
-	authState        *SSEAuthState
-	ctx              context.Context
+	mu                      sync.Mutex
+	connections             map[string]*SSEConnection // active SSE connections
+	pollingProjects         map[string]*pollingState  // projects in polling fallback
+	cache                   *Cache
+	domainURL               *url.URL
+	httpClient              *http.Client // streaming client (no timeout) for SSE connections
+	resyncHttpClient        *http.Client // regular client (with timeout) for cache resync requests
+	authState               *SSEAuthState
+	pollingFallbackInterval time.Duration
+	ctx                     context.Context
 }
 
-func NewSSEManager(ctx context.Context, cache *Cache, domainURL *url.URL, httpClient *http.Client, resyncHttpClient *http.Client, authState *SSEAuthState) *SSEManager {
-
+func NewSSEManager(ctx context.Context, cache *Cache, domainURL *url.URL, httpClient *http.Client, resyncHttpClient *http.Client, authState *SSEAuthState, pollingFallbackInterval time.Duration) *SSEManager {
 	return &SSEManager{
-		connections:      make(map[string]*SSEConnection),
-		pollingProjects:  make(map[string]*pollingState),
-		cache:            cache,
-		domainURL:        domainURL,
-		httpClient:       httpClient,
-		resyncHttpClient: resyncHttpClient,
-		authState:        authState,
-		ctx:              ctx,
+		connections:             make(map[string]*SSEConnection),
+		pollingProjects:         make(map[string]*pollingState),
+		cache:                   cache,
+		domainURL:               domainURL,
+		httpClient:              httpClient,
+		resyncHttpClient:        resyncHttpClient,
+		authState:               authState,
+		pollingFallbackInterval: pollingFallbackInterval,
+		ctx:                     ctx,
 	}
 }
 
@@ -327,9 +328,6 @@ func (m *SSEManager) runConnection(conn *SSEConnection, ctx context.Context) {
 			Int("maxRetries", maxRetries).
 			Msg("SSE connection lost, resyncing cache before retrying")
 
-		// Resync project cache before reconnecting — we may have missed events during the outage
-		runProjectSecretsRefresh(m.cache, m.domainURL, m.resyncHttpClient, conn.ProjectID, conn.EnvironmentSlug)
-
 		// Exponential backoff with jitter
 		delay := baseDelay * time.Duration(math.Pow(2, float64(retries-1)))
 
@@ -345,8 +343,6 @@ func (m *SSEManager) runConnection(conn *SSEConnection, ctx context.Context) {
 	}
 }
 
-const pollingFallbackInterval = 5 * time.Minute
-
 // transitionToPolling moves a project from SSE mode to polling fallback.
 func (m *SSEManager) transitionToPolling(projectId, environmentSlug string) {
 	m.mu.Lock()
@@ -373,7 +369,7 @@ func (m *SSEManager) transitionToPolling(projectId, environmentSlug string) {
 		Msg("Starting polling fallback for project")
 
 	// Resync the project cache immediately — events might have been missed during the outage
-	go runProjectSecretsRefresh(m.cache, m.domainURL, m.resyncHttpClient, projectId, environmentSlug)
+	runProjectSecretsRefresh(m.cache, m.domainURL, m.resyncHttpClient, projectId, environmentSlug)
 
 	// On each polling tick (except the first), attempt SSE reconnection
 	retrySSE := func() {
@@ -388,7 +384,7 @@ func (m *SSEManager) transitionToPolling(projectId, environmentSlug string) {
 		go m.attemptSSEReconnection(projectId, environmentSlug)
 	}
 
-	go startProjectPollingLoop(pollCtx, m.cache, m.domainURL, m.resyncHttpClient, projectId, environmentSlug, pollingFallbackInterval, retrySSE)
+	go startProjectPollingLoop(pollCtx, m.cache, m.domainURL, m.resyncHttpClient, projectId, environmentSlug, m.pollingFallbackInterval, retrySSE)
 }
 
 func (m *SSEManager) cancelPollingIfActive(projectId string) {