fix(sync): prevent sync race condition with replace option

pkg/sync/sync_context.go

@@ -1192,6 +1192,9 @@ func (sc *syncContext) applyObject(t *syncTask, dryRun, validate bool) (common.R
 			}
 		} else {
 			message, err = sc.resourceOps.CreateResource(context.TODO(), t.targetObj, dryRunStrategy, validate)
+			if err == nil {
+				t.operationState = common.OperationSucceeded
+			}
 		}
 	} else {
 		message, err = sc.resourceOps.ApplyResource(context.TODO(), t.targetObj, dryRunStrategy, force, validate, serverSideApply, sc.serverSideApplyManager)
@@ -1479,9 +1482,12 @@ func (sc *syncContext) processCreateTasks(state runState, tasks syncTasks, dryRu
 			}
 			if !dryRun || sc.dryRun || result == common.ResultCodeSyncFailed {
 				phase := operationPhases[result]
+				if t.operationState == common.OperationSucceeded {
+					phase = common.OperationSucceeded
+				}
 				// no resources are created in dry-run, so running phase means validation was
 				// successful and sync operation succeeded
-				if sc.dryRun && phase == common.OperationRunning {
+				if sc.dryRun && phase == common.OperationSucceeded {
 					phase = common.OperationSucceeded
 				}
 				sc.setResourceResult(t, result, phase, message)

pkg/sync/sync_context_test.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/client-go/rest"
 	testcore "k8s.io/client-go/testing"
 	"k8s.io/klog/v2/textlogger"
+	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
 	"github.com/argoproj/gitops-engine/pkg/diff"
 	"github.com/argoproj/gitops-engine/pkg/health"
@@ -296,6 +297,36 @@ func TestSyncCustomResources(t *testing.T) {
 	}
 }
 
+func TestSync_ReplaceMissingResourceRaceCondition(t *testing.T) {
+	syncCtx := newTestSyncCtx(nil, WithReplace(true))
+	syncCtx.dryRun = true
+	pod := testingutils.NewPod()
+	pod.SetNamespace(testingutils.FakeArgoCDNamespace)
+
+	syncCtx.resources = groupResources(ReconciliationResult{
+		Live:   []*unstructured.Unstructured{nil},
+		Target: []*unstructured.Unstructured{pod},
+	})
+	resourceOps, _ := syncCtx.resourceOps.(*kubetest.MockResourceOps)
+	createCount := 0
+	resourceOps.CreateResourceFunc = func(ctx context.Context, obj *unstructured.Unstructured, dryRunStrategy cmdutil.DryRunStrategy, validate bool) (string, error) {
+		createCount++
+		return "created", nil
+	}
+	resourceOps.ApplyResourceFunc = func(ctx context.Context, obj *unstructured.Unstructured, dryRunStrategy cmdutil.DryRunStrategy, force, validate, serverSideApply bool, manager string) (string, error) {
+		return "applied", nil
+	}
+
+	// First sync, resource is created
+	syncCtx.Sync()
+	phase, _, results := syncCtx.GetState()
+	assert.Equal(t, synccommon.OperationSucceeded, phase)
+	assert.Len(t, results, 1)
+	assert.Equal(t, synccommon.ResultCodeSynced, results[0].Status)
+	assert.Equal(t, synccommon.OperationSucceeded, results[0].HookPhase)
+	assert.Equal(t, 1, createCount)
+}
+
 func TestSyncSuccessfully(t *testing.T) {
 	syncCtx := newTestSyncCtx(nil, WithOperationSettings(false, true, false, false))
 	pod := testingutils.NewPod()

pkg/utils/kube/kubetest/mock_resource_operations.go

@@ -28,6 +28,9 @@ type MockResourceOps struct {
 	recordLock sync.RWMutex
 
 	getResourceFunc *func(ctx context.Context, config *rest.Config, gvk schema.GroupVersionKind, name string, namespace string) (*unstructured.Unstructured, error)
+
+	CreateResourceFunc func(ctx context.Context, obj *unstructured.Unstructured, dryRunStrategy cmdutil.DryRunStrategy, validate bool) (string, error)
+	ApplyResourceFunc  func(ctx context.Context, obj *unstructured.Unstructured, dryRunStrategy cmdutil.DryRunStrategy, force, validate, serverSideApply bool, manager string) (string, error)
 }
 
 // WithGetResourceFunc overrides the default ConvertToVersion behavior.
@@ -106,7 +109,10 @@ func (r *MockResourceOps) GetLastResourceCommand(key kube.ResourceKey) string {
 	return r.lastCommandPerResource[key]
 }
 
-func (r *MockResourceOps) ApplyResource(_ context.Context, obj *unstructured.Unstructured, _ cmdutil.DryRunStrategy, force, validate, serverSideApply bool, manager string) (string, error) {
+func (r *MockResourceOps) ApplyResource(ctx context.Context, obj *unstructured.Unstructured, dryRunStrategy cmdutil.DryRunStrategy, force, validate, serverSideApply bool, manager string) (string, error) {
+	if r.ApplyResourceFunc != nil {
+		return r.ApplyResourceFunc(ctx, obj, dryRunStrategy, force, validate, serverSideApply, manager)
+	}
 	r.SetLastValidate(validate)
 	r.SetLastServerSideApply(serverSideApply)
 	r.SetLastServerSideApplyManager(manager)
@@ -140,7 +146,10 @@ func (r *MockResourceOps) UpdateResource(_ context.Context, obj *unstructured.Un
 	return obj, command.Err
 }
 
-func (r *MockResourceOps) CreateResource(_ context.Context, obj *unstructured.Unstructured, _ cmdutil.DryRunStrategy, _ bool) (string, error) {
+func (r *MockResourceOps) CreateResource(ctx context.Context, obj *unstructured.Unstructured, dryRunStrategy cmdutil.DryRunStrategy, validate bool) (string, error) {
+	if r.CreateResourceFunc != nil {
+		return r.CreateResourceFunc(ctx, obj, dryRunStrategy, validate)
+	}
 	r.SetLastResourceCommand(kube.GetResourceKey(obj), "create")
 	command, ok := r.Commands[obj.GetName()]
 	if !ok {