server: add custom timeouts to actions (hasura#5762)

hasura#5762

CHANGELOG.md

@@ -40,6 +40,7 @@ The corresponding JWT config can be:
 - 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: change `created_at` column type from `timestamp` to `timestamptz` for scheduled triggers tables (fix #5722)
+- server: allow configuring timeouts for actions (fixes #4966)
 
 ### Breaking change
 

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

@@ -46,7 +46,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"
       }
@@ -119,6 +120,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/Resolve/Action.hs

@@ -191,7 +191,7 @@ resolveActionMutationSync env field executionContext userInfo = do
   manager <- asks getter
   reqHeaders <- asks getter
   (webhookRes, respHeaders) <- callWebhook env manager outputType outputFields reqHeaders confHeaders
-                               forwardClientHeaders resolvedWebhook handlerPayload
+                               forwardClientHeaders resolvedWebhook handlerPayload timeout
   let webhookResponseExpression = RS.AEInput $ UVSQL $
         toTxtValue $ WithScalarType PGJSONB $ PGValJSONB $ Q.JSONB $ J.toJSON webhookRes
   selSet <- asObjectSelectionSet $ _fSelSet field
@@ -212,7 +212,7 @@ resolveActionMutationSync env field executionContext userInfo = do
 
   where
     ActionExecutionContext actionName outputType outputFields definitionList resolvedWebhook confHeaders
-      forwardClientHeaders = executionContext
+      forwardClientHeaders timeout = executionContext
 
 -- QueryActionExecuter is a type for a higher function, this is being used
 -- to allow or disallow where a query action can be executed. We would like
@@ -256,7 +256,7 @@ resolveActionQuery env field executionContext sessionVariables httpManager reqHe
       actionContext = ActionContext actionName
       handlerPayload = ActionWebhookPayload actionContext sessionVariables inputArgs
   (webhookRes, _) <- callWebhook env httpManager outputType outputFields reqHeaders confHeaders
-                               forwardClientHeaders resolvedWebhook handlerPayload
+                               forwardClientHeaders resolvedWebhook handlerPayload timeout
   let webhookResponseExpression = RS.AEInput $ UVSQL $
         toTxtValue $ WithScalarType PGJSONB $ PGValJSONB $ Q.JSONB $ J.toJSON webhookRes
   selSet <- asObjectSelectionSet $ _fSelSet field
@@ -266,7 +266,7 @@ resolveActionQuery env field executionContext sessionVariables httpManager reqHe
   return selectAstUnresolved
   where
     ActionExecutionContext actionName outputType outputFields definitionList resolvedWebhook confHeaders
-      forwardClientHeaders = executionContext
+      forwardClientHeaders timeout = executionContext
 
 {- Note: [Async action architecture]
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -439,12 +439,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
@@ -516,9 +518,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")
@@ -529,11 +532,13 @@ callWebhook env manager outputType outputFields reqHeaders confHeaders
       requestBody = J.encode postPayload
       requestBodySize = BL.length requestBody
       url = unResolvedWebhook resolvedWebhook
-  httpResponse <-  do
+      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
@@ -645,4 +650,3 @@ processOutputSelectionSet tableRowInput actionOutputType definitionList fldTy fl
 
     functionArgs = RS.FunctionArgsExp [tableRowInput] mempty
     selectFrom = RS.FromFunction jsonbToPostgresRecordFunction functionArgs $ Just definitionList
-

server/src-lib/Hasura/GraphQL/Resolve/Types.hs

@@ -123,6 +123,7 @@ data ActionExecutionContext
   , _saecWebhook              :: !ResolvedWebhook
   , _saecHeaders              :: ![HeaderConf]
   , _saecForwardClientHeaders :: !Bool
+  , _saecTimeout              :: !Timeout
   } deriving (Show, Eq)
 
 data ActionMutationExecutionContext

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

@@ -71,6 +71,7 @@ mkQueryActionField actionName actionInfo definitionList =
       (_adHandler definition)
       (_adHeaders definition)
       (_adForwardClientHeaders definition)
+      (_adTimeout definition)
 
     description = mkDescriptionWith (PGDescription <$> _aiComment actionInfo) $
                   "perform the action: " <>> actionName
@@ -110,6 +111,7 @@ mkMutationActionField actionName actionInfo definitionList kind =
           (_adHandler definition)
           (_adHeaders definition)
           (_adForwardClientHeaders definition)
+          (_adTimeout definition)
         ActionAsynchronous -> ActionMutationAsync
 
     description = mkDescriptionWith (PGDescription <$> _aiComment actionInfo) $

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

@@ -134,6 +134,7 @@ resolveAction env customTypes allPGScalars actionDefinition = do
          | otherwise -> pure ()
 
   -- Check if the response type is an object
+-- <<<<<<< HEAD
   outputObject <- getObjectTypeInfo responseBaseType
   resolvedDef <- traverse (resolveWebhook env) actionDefinition
   pure (resolvedDef, outputObject, reusedPGScalars)
@@ -148,6 +149,18 @@ resolveAction env customTypes allPGScalars actionDefinition = do
         throw400 NotExists $ "the type: "
         <> showNamedTy typeName <>
         " is not an object type defined in custom types"
+-- =======
+--   let outputType = unGraphQLType _adOutputType
+--       outputBaseType = G.getBaseType outputType
+--   outputObject <- onNothing (Map.lookup outputBaseType _actObjects) $
+--     throw400 NotExists $ "the type: " <> showName outputBaseType
+--     <> " is not an object type defined in custom types"
+--   resolvedWebhook <- resolveWebhook env _adHandler
+--   pure ( ActionDefinition resolvedArguments _adOutputType _adType
+--          _adHeaders _adForwardClientHeaders _adTimeout resolvedWebhook
+--        , outputObject
+--        )
+-- >>>>>>> 9d047d172... server: add custom timeouts to actions (#5762)
 
 runUpdateAction
   :: forall m. ( QErrM m , CacheRWM m, MonadTx m)

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

@@ -24,6 +24,7 @@ module Hasura.RQL.Types.Action
   , aiPgScalars
   , aiPermissions
   , aiComment
+  , defaultActionTimeoutSecs
   , ActionPermissionInfo(..)
 
   , ActionPermissionMap
@@ -101,7 +102,10 @@ data ActionDefinition a
   , _adType                 :: !ActionType
   , _adHeaders              :: ![HeaderConf]
   , _adForwardClientHeaders :: !Bool
+  , _adTimeout              :: !Timeout
   , _adHandler              :: !a
+  -- ^ If the timeout is not provided by the user, then
+  -- the default timeout of 30 seconds will be used
   } deriving (Show, Eq, Lift, Functor, Foldable, Traversable, Generic)
 instance (NFData a) => NFData (ActionDefinition a)
 instance (Cacheable a) => Cacheable (ActionDefinition a)
@@ -113,6 +117,7 @@ instance (J.FromJSON a) => J.FromJSON (ActionDefinition a) 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
@@ -121,7 +126,7 @@ instance (J.FromJSON a) => J.FromJSON (ActionDefinition a) where
     return ActionDefinition {..}
 
 instance (J.ToJSON a) => J.ToJSON (ActionDefinition a) 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)
@@ -132,6 +137,7 @@ instance (J.ToJSON a) => J.ToJSON (ActionDefinition a) where
     , "headers"                J..= headers
     , "forward_client_headers" J..= forwardClientHeaders
     , "handler"                J..= handler
+    , "timeout"                J..= timeout
     ] <> typeAndKind
 
 type ResolvedActionDefinition = ActionDefinition ResolvedWebhook

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
@@ -57,6 +60,7 @@ import           Data.Aeson.Casing
 import           Data.Bifunctor                (bimap)
 import           Data.Aeson.TH
 import           Data.Sequence.NonEmpty
+import           Data.Scientific               (toBoundedInteger)
 import           Data.URL.Template
 import           Instances.TH.Lift             ()
 import           Language.Haskell.TH.Syntax    (Lift, Q, TExp)
@@ -314,3 +318,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);