server: add custom timeouts to actions

CHANGELOG.md

@@ -55,6 +55,7 @@ This release contains the [PDV refactor (#4111)](https://github.com/hasura/graph
 - server: treat the absence of `backend_only` configuration and `backend_only: false` equally (closing #5059) (#4111)
 - server: allow remote relationships joining `type` column with `[type]` input argument as spec allows this coercion (fixes #5133)
 - server: add action-like URL templating for event triggers and remote schemas (fixes #2483)
+- server: allow adding custom timeouts to actions (fixes #4966)
 - console: allow user to cascade Postgres dependencies when dropping Postgres objects (close #5109) (#5248)
 - console: mark inconsistent remote schemas in the UI (close #5093) (#5181)
 - cli: add missing global flags for seeds command (#5565)

docs/graphql/core/api-reference/schema-metadata-api/actions.rst

@@ -49,7 +49,8 @@ Create a synchronous action with name ``create_user``:
                }
             ],
             "output_type":"User",
-            "handler":"https://action.my_app.com/create-user"
+            "handler":"https://action.my_app.com/create-user",
+            "timeout":60
          },
          "comment": "Custom action to create user"
       }
@@ -122,6 +123,11 @@ ActionDefinition
      - false
      - [ ``mutation`` | ``query`` ]
      - The type of the action (default: ``mutation``)
+   * - timeout
+     - false
+     - Integer
+     - Number of seconds to wait for response before timing out. Default: 30
+
 
 .. _InputArgument:
 

server/src-lib/Hasura/GraphQL/Execute/Action.hs

@@ -154,7 +154,7 @@ resolveActionExecution env logger userInfo annAction execContext = do
   let actionContext = ActionContext actionName
       handlerPayload = ActionWebhookPayload actionContext sessionVariables inputPayload
   (webhookRes, respHeaders) <- flip runReaderT logger $ callWebhook env manager outputType outputFields reqHeaders confHeaders
-                               forwardClientHeaders resolvedWebhook handlerPayload
+                               forwardClientHeaders resolvedWebhook handlerPayload timeout
   let webhookResponseExpression = RS.AEInput $ UVLiteral $
         toTxtValue $ WithScalarType PGJSONB $ PGValJSONB $ Q.JSONB $ J.toJSON webhookRes
       selectAstUnresolved = processOutputSelectionSet webhookResponseExpression
@@ -164,7 +164,7 @@ resolveActionExecution env logger userInfo annAction execContext = do
   where
     AnnActionExecution actionName outputType annFields inputPayload
       outputFields definitionList resolvedWebhook confHeaders
-      forwardClientHeaders stringifyNum = annAction
+      forwardClientHeaders stringifyNum timeout = annAction
     ActionExecContext manager reqHeaders sessionVariables = execContext
 
 
@@ -331,12 +331,14 @@ asyncActionsProcessor env logger cacheRef pgPool httpManager = forever $ do
               webhookUrl = _adHandler definition
               forwardClientHeaders = _adForwardClientHeaders definition
               confHeaders = _adHeaders definition
+              timeout = _adTimeout definition
               outputType = _adOutputType definition
               actionContext = ActionContext actionName
           eitherRes <- runExceptT $ flip runReaderT logger $
                        callWebhook env httpManager outputType outputFields reqHeaders confHeaders
-                         forwardClientHeaders webhookUrl $
-                         ActionWebhookPayload actionContext sessionVariables inputPayload
+                         forwardClientHeaders webhookUrl (ActionWebhookPayload actionContext sessionVariables inputPayload)
+                         timeout
+
           liftIO $ case eitherRes of
             Left e                     -> setError actionId e
             Right (responsePayload, _) -> setCompleted actionId $ J.toJSON responsePayload
@@ -408,9 +410,10 @@ callWebhook
   -> Bool
   -> ResolvedWebhook
   -> ActionWebhookPayload
+  -> Timeout
   -> m (ActionWebhookResponse, HTTP.ResponseHeaders)
 callWebhook env manager outputType outputFields reqHeaders confHeaders
-            forwardClientHeaders resolvedWebhook actionWebhookPayload = do
+            forwardClientHeaders resolvedWebhook actionWebhookPayload timeoutSeconds = do
   resolvedConfHeaders <- makeHeadersFromConf env confHeaders
   let clientHeaders = if forwardClientHeaders then mkClientHeadersForward reqHeaders else []
       contentType = ("Content-Type", "application/json")
@@ -421,11 +424,13 @@ callWebhook env manager outputType outputFields reqHeaders confHeaders
       requestBody = J.encode postPayload
       requestBodySize = BL.length requestBody
       url = unResolvedWebhook resolvedWebhook
+      responseTimeout = HTTP.responseTimeoutMicro $ (unTimeout timeoutSeconds) * 1000000
   httpResponse <- do
     initReq <- liftIO $ HTTP.parseRequest (T.unpack url)
-    let req = initReq { HTTP.method         = "POST"
-                      , HTTP.requestHeaders = addDefaultHeaders hdrs
-                      , HTTP.requestBody    = HTTP.RequestBodyLBS requestBody
+    let req = initReq { HTTP.method          = "POST"
+                      , HTTP.requestHeaders  = addDefaultHeaders hdrs
+                      , HTTP.requestBody     = HTTP.RequestBodyLBS requestBody
+                      , HTTP.responseTimeout = responseTimeout
                       }
     Tracing.tracedHttpRequest req \req' ->
       liftIO . try $ HTTP.httpLbs req' manager

server/src-lib/Hasura/GraphQL/Schema/Action.hs

@@ -62,6 +62,7 @@ actionExecute nonObjectTypeMap actionInfo = runMaybeT do
                , _aaeHeaders = _adHeaders definition
                , _aaeForwardClientHeaders = _adForwardClientHeaders definition
                , _aaeStrfyNum = stringifyNum
+               , _aaeTimeOut = _adTimeout definition
                }
   where
     ActionInfo actionName outputObject definition permissions comment = actionInfo

server/src-lib/Hasura/RQL/DDL/Action.hs

@@ -125,7 +125,7 @@ resolveAction env AnnotatedCustomTypes{..} ActionDefinition{..} allPGScalars = d
     <> " is not an object type defined in custom types"
   resolvedWebhook <- resolveWebhook env _adHandler
   pure ( ActionDefinition resolvedArguments _adOutputType _adType
-         _adHeaders _adForwardClientHeaders resolvedWebhook
+         _adHeaders _adForwardClientHeaders _adTimeout resolvedWebhook
        , outputObject
        )
 

server/src-lib/Hasura/RQL/DDL/Metadata/Types.hs

@@ -538,7 +538,8 @@ replaceMetadataToOrdJSON ( ReplaceMetadata
           <> catMaybes [maybeAnyToMaybeOrdPair "description" AO.toOrdered descM]
 
         actionDefinitionToOrdJSON :: ActionDefinitionInput -> AO.Value
-        actionDefinitionToOrdJSON (ActionDefinition args outputType actionType headers frwrdClientHdrs handler) =
+        actionDefinitionToOrdJSON (ActionDefinition args outputType actionType
+                                   headers frwrdClientHdrs timeout handler) =
           let typeAndKind = case actionType of
                 ActionQuery -> [("type", AO.toOrdered ("query" :: String))]
                 ActionMutation kind -> [ ("type", AO.toOrdered ("mutation" :: String))
@@ -551,6 +552,7 @@ replaceMetadataToOrdJSON ( ReplaceMetadata
           <> catMaybes [ listToMaybeOrdPair "headers" AO.toOrdered headers
                        , listToMaybeOrdPair "arguments" argDefinitionToOrdJSON args]
           <> typeAndKind
+          <> (bool [("timeout",AO.toOrdered timeout)] mempty $ timeout == defaultActionTimeoutSecs)
 
         permToOrdJSON :: ActionPermissionMetadata -> AO.Value
         permToOrdJSON (ActionPermissionMetadata role permComment) =

server/src-lib/Hasura/RQL/Types/Action.hs

@@ -10,9 +10,10 @@ module Hasura.RQL.Types.Action
   , adArguments
   , adOutputType
   , adType
-  , adHeaders
   , adForwardClientHeaders
+  , adHeaders
   , adHandler
+  , adTimeout
   , ActionType(..)
   , _ActionMutation
   , CreateAction(..)
@@ -31,6 +32,7 @@ module Hasura.RQL.Types.Action
   , aiDefinition
   , aiPermissions
   , aiComment
+  , defaultActionTimeoutSecs
   , ActionPermissionInfo(..)
 
   , ActionPermissionMap
@@ -122,6 +124,9 @@ data ActionDefinition a b
   , _adType                 :: !ActionType
   , _adHeaders              :: ![HeaderConf]
   , _adForwardClientHeaders :: !Bool
+  , _adTimeout              :: !Timeout
+  -- ^ If the timeout is not provided by the user, then
+  -- the default timeout of 30 seconds will be used
   , _adHandler              :: !b
   } deriving (Show, Eq, Lift, Functor, Foldable, Traversable, Generic)
 instance (NFData a, NFData b) => NFData (ActionDefinition a b)
@@ -135,6 +140,7 @@ instance (J.FromJSON a, J.FromJSON b) => J.FromJSON (ActionDefinition a b) where
     _adHeaders <- o J..:? "headers" J..!= []
     _adForwardClientHeaders <- o J..:? "forward_client_headers" J..!= False
     _adHandler <- o J..:  "handler"
+    _adTimeout <- o J..:? "timeout" J..!= defaultActionTimeoutSecs
     actionType <- o J..:? "type" J..!= "mutation"
     _adType <- case actionType of
       "mutation" -> ActionMutation <$> o J..:? "kind" J..!= ActionSynchronous
@@ -143,7 +149,7 @@ instance (J.FromJSON a, J.FromJSON b) => J.FromJSON (ActionDefinition a b) where
     return ActionDefinition {..}
 
 instance (J.ToJSON a, J.ToJSON b) => J.ToJSON (ActionDefinition a b) where
-  toJSON (ActionDefinition args outputType actionType headers forwardClientHeaders handler) =
+  toJSON (ActionDefinition args outputType actionType headers forwardClientHeaders timeout handler) =
     let typeAndKind = case actionType of
           ActionQuery -> [ "type" J..= ("query" :: String)]
           ActionMutation kind -> [ "type" J..= ("mutation" :: String)
@@ -154,6 +160,7 @@ instance (J.ToJSON a, J.ToJSON b) => J.ToJSON (ActionDefinition a b) where
     , "headers"                J..= headers
     , "forward_client_headers" J..= forwardClientHeaders
     , "handler"                J..= handler
+    , "timeout"                J..= timeout
     ] <> typeAndKind
 
 type ResolvedActionDefinition =
@@ -262,6 +269,7 @@ data AnnActionExecution v
   , _aaeHeaders              :: ![HeaderConf]
   , _aaeForwardClientHeaders :: !Bool
   , _aaeStrfyNum             :: !Bool
+  , _aaeTimeOut              :: !Timeout
   } deriving (Show, Eq)
 
 data AnnActionMutationAsync

server/src-lib/Hasura/RQL/Types/Common.hs

@@ -41,6 +41,9 @@ module Hasura.RQL.Types.Common
        , InputWebhook(..)
        , ResolvedWebhook(..)
        , resolveWebhook
+
+       , Timeout(..)
+       , defaultActionTimeoutSecs
        ) where
 
 import           Hasura.EncJSON
@@ -58,6 +61,7 @@ import           Data.Aeson
 import           Data.Aeson.Casing
 import           Data.Bifunctor                (bimap)
 import           Data.Aeson.TH
+import           Data.Scientific               (toBoundedInteger)
 import           Data.URL.Template
 import           Instances.TH.Lift             ()
 import           Language.Haskell.TH.Syntax    (Lift, Q, TExp)
@@ -316,3 +320,19 @@ resolveWebhook env (InputWebhook urlTemplate) = do
   let eitherRenderedTemplate = renderURLTemplate env urlTemplate
   either (throw400 Unexpected . T.pack)
     (pure . ResolvedWebhook) eitherRenderedTemplate
+
+newtype Timeout = Timeout { unTimeout :: Int }
+  deriving (Show, Eq, ToJSON, Generic, NFData, Cacheable, Lift)
+
+instance FromJSON Timeout where
+  parseJSON = withScientific "Timeout" $ \t -> do
+    timeout <- onNothing (toBoundedInteger t) $ fail (show t <> " is out of bounds")
+    case (timeout >= 0) of
+      True  -> return $ Timeout timeout
+      False -> fail "timeout value cannot be negative"
+
+instance Arbitrary Timeout where
+  arbitrary = Timeout <$> QC.choose (0, 10000000)
+
+defaultActionTimeoutSecs :: Timeout
+defaultActionTimeoutSecs = Timeout 30

server/tests-py/context.py

@@ -179,6 +179,11 @@ def do_POST(self):
             resp, status = self.create_user()
             self._send_response(status, resp)
 
+        elif req_path == "/create-user-timeout":
+            time.sleep(2)
+            resp, status = self.create_user()
+            self._send_response(status, resp)
+
         elif req_path == "/create-users":
             resp, status = self.create_users()
             self._send_response(status, resp)

server/tests-py/queries/actions/timeout/schema_setup.yaml

@@ -0,0 +1,44 @@
+type: bulk
+args:
+- type: run_sql
+  args:
+    sql: |
+      CREATE TABLE "user"(
+        id SERIAL PRIMARY KEY,
+        name TEXT NOT NULL,
+        email TEXT NOT NULL,
+        is_admin BOOLEAN NOT NULL DEFAULT false
+      );
+
+- type: track_table
+  args:
+    name: user
+    schema: public
+
+- type: set_custom_types
+  args:
+    objects:
+    - name: UserId
+      fields:
+      - name: id
+        type: Int!
+      relationships:
+      - name: user
+        type: object
+        remote_table: user
+        field_mapping:
+          id: id
+
+- type: create_action
+  args:
+    name: create_user
+    definition:
+      kind: asynchronous
+      arguments:
+      - name: email
+        type: String!
+      - name: name
+        type: String!
+      output_type: UserId
+      timeout: 2
+      handler: http://127.0.0.1:5593/create-user-timeout

server/tests-py/queries/actions/timeout/schema_teardown.yaml

@@ -0,0 +1,15 @@
+type: bulk
+args:
+- type: drop_action
+  args:
+    name: create_user
+    clear_data: true
+# clear custom types
+- type: set_custom_types
+  args: {}
+
+- type: run_sql
+  args:
+    cascade: true
+    sql: |
+      DROP TABLE "user";

server/tests-py/queries/actions/timeout/values_teardown.yaml

@@ -0,0 +1,7 @@
+type: bulk
+args:
+- type: run_sql
+  args:
+    sql: |
+      DELETE FROM "user";
+      SELECT setval('user_id_seq', 1, FALSE);

server/tests-py/test_actions.py

@@ -475,3 +475,50 @@ def test_introspection_query(self, hge_ctx):
         code, resp, _ = hge_ctx.anyq(conf['url'], conf['query'], headers)
         assert code == 200, resp
         assert 'data' in resp, resp
+
+@use_action_fixtures
+class TestActionTimeout:
+
+    @classmethod
+    def dir(cls):
+        return 'queries/actions/timeout'
+
+    def test_action_timeout_fail(self, hge_ctx):
+        graphql_mutation = '''
+        mutation {
+          create_user(email: "random-email", name: "Clarke")
+        }
+        '''
+        query = {
+            'query': graphql_mutation,
+            'variables': {}
+        }
+        status, resp, _ = hge_ctx.anyq('/v1/graphql', query, mk_headers_with_secret(hge_ctx))
+        assert status == 200, resp
+        assert 'data' in resp
+        action_id = resp['data']['create_user']
+        query_async = '''
+        query ($action_id: uuid!){
+          create_user(id: $action_id){
+            id
+            errors
+          }
+        }
+        '''
+        query = {
+            'query': query_async,
+            'variables': {
+                'action_id': action_id
+            }
+        }
+        conf = {
+            'url': '/v1/graphql',
+            'headers': {},
+            'query': query,
+            'status': 200,
+        }
+        # the action takes 2 seconds to complete
+        time.sleep(3)
+        response, _ = check_query(hge_ctx, conf)
+        assert 'errors' in response['data']['create_user']
+        assert 'ResponseTimeout' == response['data']['create_user']['errors']['internal']['error']['message']