Fix locking

Author: meghan-bailey

flow/process_genericflow.go

@@ -274,6 +274,12 @@ func ProcessTableConfigurations(tfmContext FlowMachineContext, tfContext FlowCon
 }
 
 func ProcessFlowStatesForInterval(tfContext FlowContext, tfmContext FlowMachineContext, flowDefinitionContext *FlowLibraryContext, regionList []string) int {
+	// Acquire the pullOnceMu so that synchronous ExecuteFilteredPullOnce calls block
+	// all async push/pull activity for this flow until they complete.
+	mu := tfContext.GetPullOnceMu()
+	mu.Lock()
+	defer mu.Unlock()
+
 	if tfContext.GetFlowStateState() == 3 {
 		tfContext.SetRestart(false)
 		tfmContext.SetPermissionUpdate(tfContext)
@@ -331,12 +337,6 @@ func ProcessFlowStatesForInterval(tfContext FlowContext, tfmContext FlowMachineC
 		return 0
 	}
 
-	// Acquire the pullOnceMu so that synchronous ExecuteFilteredPullOnce calls block
-	// all async push/pull activity for this flow until they complete.
-	mu := tfContext.GetPullOnceMu()
-	mu.Lock()
-	defer mu.Unlock()
-
 	// Logic for push/pull once
 	if tfContext.FlowSyncModeMatch("push", true) {
 		switch syncSuffix := strings.TrimPrefix(tfContext.GetFlowSyncMode(), "push"); syncSuffix {
@@ -484,6 +484,14 @@ func ExecuteFilteredPullOnce(tfmContext FlowMachineContext, tfContext FlowContex
 		return
 	}
 
+	// Acquire the mutex before mutating state so the async path cannot race with us.
+	mu := tfContext.GetPullOnceMu()
+	mu.Lock()
+	defer mu.Unlock()
+
+	tfContext.SetFlowSyncFilter(syncFilter)
+	tfContext.SetFlowSyncMode("pullonce")
+
 	var tableIndexKey string
 	if flowDefinitionContext.GetTableIndexColumnNames != nil {
 		if keys := flowDefinitionContext.GetTableIndexColumnNames(); len(keys) == 1 {
@@ -495,14 +503,6 @@ func ExecuteFilteredPullOnce(tfmContext FlowMachineContext, tfContext FlowContex
 		return
 	}
 
-	// Acquire the mutex before mutating state so the async path cannot race with us.
-	mu := tfContext.GetPullOnceMu()
-	mu.Lock()
-	defer mu.Unlock()
-
-	tfContext.SetFlowSyncFilter(syncFilter)
-	tfContext.SetFlowSyncMode("pullonce")
-
 	tableConfigurations, err := tableConfigurationFlowPullRemote(tfmContext, tfContext)
 	if err != nil {
 		tfmContext.Log("ExecuteFilteredPullOnce: error pulling configurations", err)
@@ -533,6 +533,7 @@ func ExecuteFilteredPullOnce(tfmContext FlowMachineContext, tfContext FlowContex
 		}
 	}
 
+	tfContext.SetFlowSyncFilter("")
 	tfContext.SetFlowSyncMode("pullcomplete")
 	tfContext.PushState("flowStateReceiver", tfContext.NewFlowStateUpdate("2", "pullcomplete"))
 }