jason810496 commented on code in PR #75601:
URL: https://github.com/apache/airflow/pull/75601#discussion_r4435428119
##########
go-sdk/pkg/execution/client.go:
##########
@@ -529,26 +540,47 @@ func resolveExpiry(retention *time.Duration, now
time.Time) (any, error) {
// validateJSONRepresentable checks what the frame encoder actually emits, not
// the Go value: a reflection walk has to mirror the encoder's field rules
(tags,
// "-", omitempty, embedding, marshalers) and misjudges values wherever it
drifts.
-func validateJSONRepresentable(value any) error {
+// On success it also returns the decoded value, which is nil when value
encodes
+// to null.
+func validateJSONRepresentable(value any) (any, error) {
var buf bytes.Buffer
if err := newFrameEncoder(&buf).Encode(value); err != nil {
- return fmt.Errorf("%T is not JSON representable: %w", value,
err)
+ return nil, fmt.Errorf("%T is not JSON representable: %w",
value, err)
}
dec := msgpack.NewDecoder(&buf)
- // The default map decoder fails opaquely on non-string keys.
- dec.SetMapDecoder(func(d *msgpack.Decoder) (any, error) {
- return d.DecodeUntypedMap()
- })
+ dec.SetMapDecoder(decodeStringKeyedMap)
decoded, err := dec.DecodeInterface()
if err != nil {
- return fmt.Errorf("%T is not JSON representable: %w", value,
err)
+ return nil, fmt.Errorf("%T is not JSON representable: %w",
value, err)
}
- // Checked after decoding: a typed nil such as a nil *string is not ==
nil,
- // yet it still encodes to null, which the Execution API rejects.
- if decoded == nil {
- return errors.New("value must not be nil")
+ if err := validateDecodedValue(decoded); err != nil {
+ return nil, err
}
- return validateDecodedValue(decoded)
+ return decoded, nil
+}
Review Comment:
Return the encoded bytes so the caller sends them instead of encoding again.
```suggestion
// validateJSONRepresentable checks what the frame encoder actually emits,
not
// the Go value: a reflection walk has to mirror the encoder's field rules
(tags,
// "-", omitempty, embedding, marshalers) and misjudges values wherever it
drifts.
// On success it returns the encoded bytes, so the caller sends them instead
of
// encoding the value a second time, and whether the value encodes to null.
func validateJSONRepresentable(value any) (msgpack.RawMessage, bool, error) {
var buf bytes.Buffer
if err := newFrameEncoder(&buf).Encode(value); err != nil {
return nil, false, fmt.Errorf("%T is not JSON representable:
%w", value, err)
}
encoded := msgpack.RawMessage(buf.Bytes())
// Decoded from its own reader so encoded keeps the whole payload.
dec := msgpack.NewDecoder(bytes.NewReader(encoded))
dec.SetMapDecoder(decodeStringKeyedMap)
decoded, err := dec.DecodeInterface()
if err != nil {
return nil, false, fmt.Errorf("%T is not JSON representable:
%w", value, err)
}
if err := validateDecodedValue(decoded); err != nil {
return nil, false, err
}
return encoded, decoded == nil, nil
}
```
##########
go-sdk/pkg/execution/client.go:
##########
@@ -315,6 +315,11 @@ func (c *CoordinatorClient) PushXCom(
key string,
value any,
) error {
+ // Unlike a task state value, nil is allowed: it is pushed as JSON null.
+ if _, err := validateJSONRepresentable(value); err != nil {
+ return fmt.Errorf("cannot push XCom %q: %w", key, err)
+ }
+
msg := genmodels.SetXCom{
Key: key,
Value: value,
Review Comment:
Send the bytes validation already produced.
```suggestion
// Unlike a task state value, nil is allowed: it is pushed as JSON null.
encoded, _, err := validateJSONRepresentable(value)
if err != nil {
return fmt.Errorf("cannot push XCom %q: %w", key, err)
}
msg := genmodels.SetXCom{
Key: key,
Value: encoded,
```
##########
go-sdk/pkg/execution/client.go:
##########
@@ -468,9 +473,15 @@ func (s taskStateStore) Set(
// The caller's value is checked first so a misconfigured deployment
cannot
// mask a programming error in the task.
- if err := validateJSONRepresentable(value); err != nil {
+ decoded, err := validateJSONRepresentable(value)
+ if err != nil {
return fmt.Errorf("cannot set task state key %q: %w", key, err)
}
+ // Checked after decoding: a typed nil such as a nil *string is not ==
nil,
+ // yet it still encodes to null, which the Execution API rejects.
+ if decoded == nil {
+ return fmt.Errorf("cannot set task state key %q: value must not
be nil", key)
+ }
Review Comment:
Same here, plus `Value: encoded` in the literal below (line 494).
```suggestion
// The caller's value is checked first so a misconfigured deployment
cannot
// mask a programming error in the task.
encoded, isNull, err := validateJSONRepresentable(value)
if err != nil {
return fmt.Errorf("cannot set task state key %q: %w", key, err)
}
// Checked after encoding: a typed nil such as a nil *string is not ==
nil,
// yet it still encodes to null, which the Execution API rejects.
if isNull {
return fmt.Errorf("cannot set task state key %q: value must not
be nil", key)
}
```
##########
go-sdk/sdk/sdk.go:
##########
@@ -126,6 +126,10 @@ type XComClient interface {
) (any, error)
// PushXCom stores value under key for the given task instance ti.
+ //
+ // value must be nil or built from strings, numbers, bools, slices,
+ // string-keyed maps, and structs. A time.Time, a byte slice or array
(such
+ // as a uuid.UUID), or a non-finite float is rejected before it is sent.
Review Comment:
Shorter; the list of allowed types goes stale.
```suggestion
// value must be JSON serializable; nil is allowed and is pushed as
null.
// A time.Time, a byte slice or array such as a uuid.UUID, or a
non-finite
// float is rejected before it is sent.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
