Handle 409s from /step endpoint in Metaflow Service (#258)

With AWS Step Functions, the metaflow task runtime registers
the step metadata, which is racy. This PR addresses that by properly
handling the 409 response from the metaflow service.

Author: savingoyal

metaflow/metadata/service.py

@@ -200,7 +200,7 @@ def create_object():
                 raise
 
     @classmethod
-    def _request(cls, monitor, path, data=None):
+    def _request(cls, monitor, path, data=None, retry_409_path=None):
         if cls.INFO is None:
             raise MetaflowException('Missing Metaflow Service URL. '
                 'Specify with METAFLOW_SERVICE_URL environment variable')
@@ -231,6 +231,19 @@ def _request(cls, monitor, path, data=None):
             else:
                 if resp.status_code < 300:
                     return resp.json()
+                elif resp.status_code == 409 and data is not None:
+                    # a special case: the post fails due to a conflict
+                    # this could occur when we missed a success response
+                    # from the first POST request but the request
+                    # actually went though, so a subsequent POST
+                    # returns 409 (conflict) or we end up with a
+                    # conflict while running on AWS Step Functions
+                    # instead of retrying the post we retry with a get since
+                    # the record is guaranteed to exist
+                    if retry_409_path:
+                        return self._request(retry_409_path)
+                    else:
+                        return
                 elif resp.status_code != 503:
                     raise ServiceException('Metadata request (%s) failed (code %s): %s'
                                            % (path, resp.status_code, resp.text),