fix: prevent 'argocd app sync' hangs if sync is completed too quickly (…

…argoproj#4373)

common/common.go

@@ -152,6 +152,8 @@ const (
 	EnvK8sClientMaxIdleConnections = "ARGOCD_K8S_CLIENT_MAX_IDLE_CONNECTIONS"
 	// EnvGnuPGHome is the path to ArgoCD's GnuPG keyring for signature verification
 	EnvGnuPGHome = "ARGOCD_GNUPGHOME"
+	// EnvWatchAPIBufferSize is the buffer size used to transfer K8S watch events to watch API consumer
+	EnvWatchAPIBufferSize = "ARGOCD_WATCH_API_BUFFER_SIZE"
 )
 
 const (

server/application/application.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	goio "io"
+	"math"
 	"reflect"
 	"sort"
 	"strconv"
@@ -49,6 +50,7 @@ import (
 	"github.com/argoproj/argo-cd/util/argo"
 	argoutil "github.com/argoproj/argo-cd/util/argo"
 	"github.com/argoproj/argo-cd/util/db"
+	"github.com/argoproj/argo-cd/util/env"
 	"github.com/argoproj/argo-cd/util/git"
 	"github.com/argoproj/argo-cd/util/helm"
 	"github.com/argoproj/argo-cd/util/lua"
@@ -57,6 +59,10 @@ import (
 	"github.com/argoproj/argo-cd/util/settings"
 )
 
+var (
+	watchAPIBufferSize = env.ParseNumFromEnv(argocommon.EnvWatchAPIBufferSize, 1000, 0, math.MaxInt32)
+)
+
 // Server provides a Application service
 type Server struct {
 	ns             string
@@ -319,7 +325,7 @@ func (s *Server) Get(ctx context.Context, q *application.ApplicationQuery) (*app
 	appIf := s.appclientset.ArgoprojV1alpha1().Applications(s.ns)
 
 	// subscribe early with buffered channel to ensure we don't miss events
-	events := make(chan *appv1.ApplicationWatchEvent, 100)
+	events := make(chan *appv1.ApplicationWatchEvent, watchAPIBufferSize)
 	unsubscribe := s.appBroadcaster.Subscribe(events, func(event *appv1.ApplicationWatchEvent) bool {
 		return event.Application.Name == q.GetName()
 	})
@@ -669,7 +675,7 @@ func (s *Server) Watch(q *application.ApplicationQuery, ws application.Applicati
 		}
 	}
 
-	events := make(chan *appv1.ApplicationWatchEvent)
+	events := make(chan *appv1.ApplicationWatchEvent, watchAPIBufferSize)
 	// Mimic watch API behavior: send ADDED events if no resource version provided
 	// If watch API is executed for one application when emit event even if resource version is provided
 	// This is required since single app watch API is used for during operations like app syncing and it is

server/application/broadcaster.go

@@ -42,6 +42,9 @@ func (b *broadcasterHandler) notify(event *appv1.ApplicationWatchEvent) {
 	}
 }
 
+// Subscribe forward application informer watch events to the provided channel.
+// The watch events are dropped if no receives are reading events from the channel so the channel must have
+// buffer if dropping events is not acceptable.
 func (b *broadcasterHandler) Subscribe(ch chan *appv1.ApplicationWatchEvent, filters ...func(event *appv1.ApplicationWatchEvent) bool) func() {
 	b.lock.Lock()
 	defer b.lock.Unlock()