@@ -128,6 +128,40 @@ pub(super) async fn decode(
128128 . map_err ( |_| refuse ( StatusCode :: SERVICE_UNAVAILABLE , "query_codec_unavailable" ) ) ?
129129}
130130
131+ /// A query's arguments as the server reads them: a list of values, or why
132+ /// they are not one. Only a declared query contract reads them
133+ /// (`validateContractedQueryArguments`); otherwise the worker receives the
134+ /// arguments as given. A map is refused as not a list, although PHP decodes
135+ /// a string-keyed map to an array and normalizes it as named arguments
136+ /// (`WorkflowQueryContract::normalizeNamedArguments`); the port's gap list
137+ /// records that.
138+ pub ( super ) enum QueryArguments {
139+ Decoded ( Vec < crate :: codec:: Value > ) ,
140+ Undecodable ,
141+ NotAList ,
142+ }
143+
144+ /// Decodes a query's arguments without refusing them.
145+ pub ( super ) async fn read_arguments ( semaphore : Arc < Semaphore > , blob : String ) -> Result < QueryArguments > {
146+ let permit = semaphore
147+ . acquire_owned ( )
148+ . await
149+ . map_err ( |_| refuse ( StatusCode :: SERVICE_UNAVAILABLE , "query_codec_unavailable" ) ) ?;
150+ tokio:: task:: spawn_blocking ( move || {
151+ let _permit = permit;
152+ let Ok ( codec) = crate :: codec:: ValueCodec :: new ( ) else {
153+ return QueryArguments :: Undecodable ;
154+ } ;
155+ match codec. decode ( & blob) {
156+ Ok ( crate :: codec:: Value :: Array ( values) ) => QueryArguments :: Decoded ( values) ,
157+ Ok ( _) => QueryArguments :: NotAList ,
158+ Err ( _) => QueryArguments :: Undecodable ,
159+ }
160+ } )
161+ . await
162+ . map_err ( |_| refuse ( StatusCode :: SERVICE_UNAVAILABLE , "query_codec_unavailable" ) )
163+ }
164+
131165/// One admission attempt: an answer, or no worker to route the query to.
132166pub ( super ) enum QueryAttempt {
133167 Answered ( ( StatusCode , Value ) ) ,
@@ -179,14 +213,15 @@ fn admits_queries(status: &str) -> bool {
179213}
180214
181215/// `validateContractedQueryArguments`: a run that declares no queries leaves
182- /// the name and arguments to the worker. A declared query without a
183- /// parameter contract takes its arguments as given.
216+ /// the name and arguments to the worker. A declared query's arguments must
217+ /// decode to a list, and its parameter contract, when it has one, checks
218+ /// them.
184219fn check_query_contract (
185220 contract : & Value ,
186221 workflow_id : & str ,
187222 run_id : & str ,
188223 name : & str ,
189- values : & [ crate :: codec :: Value ] ,
224+ arguments : & QueryArguments ,
190225) -> Result < ( ) > {
191226 let declared = |field : & str | contract[ field] . as_array ( ) . is_some_and ( |items| !items. is_empty ( ) ) ;
192227 if !declared ( "declared_queries" ) && !declared ( "declared_query_contracts" ) {
@@ -196,6 +231,22 @@ fn check_query_contract(
196231 return Err ( query_refusal ( StatusCode :: NOT_FOUND , workflow_id, run_id, name, "rejected_unknown_query" ,
197232 & format ! ( "Workflow query [{name}] is not declared on workflow [{workflow_id}]." ) ) ) ;
198233 }
234+ // A declared query's arguments must decode to a list before routing.
235+ let ( message, error) = match arguments {
236+ QueryArguments :: Decoded ( values) => return check_declared_arguments ( contract, name, values) ,
237+ QueryArguments :: Undecodable => ( "could not be decoded" , "Query arguments could not be decoded." ) ,
238+ QueryArguments :: NotAList => ( "must decode to an array" , "Query arguments must decode to an array." ) ,
239+ } ;
240+ let mut refusal = query_refusal ( StatusCode :: UNPROCESSABLE_ENTITY , workflow_id, run_id, name, "invalid_query_arguments" ,
241+ & format ! ( "Workflow query [{name}] arguments {message}." ) ) ;
242+ if let super :: RuntimeError :: Protocol { response, .. } = & mut refusal {
243+ response[ "validation_errors" ] = json ! ( { "arguments" : [ error] } ) ;
244+ }
245+ Err ( refusal)
246+ }
247+
248+ /// A declared query's parameter contract, when it has one.
249+ fn check_declared_arguments ( contract : & Value , name : & str , values : & [ crate :: codec:: Value ] ) -> Result < ( ) > {
199250 if contract[ "declared_query_contracts" ]
200251 . as_array ( )
201252 . is_some_and ( |contracts| contracts. iter ( ) . any ( |entry| entry[ "name" ] == name) )
@@ -298,7 +349,7 @@ where
298349 selected_run : Option < & str > ,
299350 name : & str ,
300351 arguments : String ,
301- values : Vec < crate :: codec :: Value > ,
352+ values : QueryArguments ,
302353 config : & QueryTaskConfig ,
303354 command : Option < & Value > ,
304355 ) -> Result < ( StatusCode , Value ) > {
@@ -328,7 +379,7 @@ where
328379 selected_run : Option < & str > ,
329380 name : & str ,
330381 arguments : & str ,
331- values : & [ crate :: codec :: Value ] ,
382+ values : & QueryArguments ,
332383 config : & QueryTaskConfig ,
333384 command : Option < & Value > ,
334385 ) -> Result < QueryAttempt > {
0 commit comments