diff --git a/ldclient/impl/datasourcev2/polling_common.py b/ldclient/impl/datasourcev2/polling_common.py index 532efd9b..b7f9bc28 100644 --- a/ldclient/impl/datasourcev2/polling_common.py +++ b/ldclient/impl/datasourcev2/polling_common.py @@ -11,7 +11,8 @@ from ldclient.impl.datasystem.protocolv2 import ( DeleteObject, EventName, - PutObject + PutObject, + has_unsupported_kind ) from ldclient.impl.util import ( _LD_ENVID_HEADER, @@ -68,6 +69,10 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: builder.start(server_intent.payload.code) elif event["event"] == EventName.PUT_OBJECT: + if has_unsupported_kind(event["data"]): + log.debug("Ignoring put-object with unsupported kind: %s", event["data"].get("kind")) + continue + try: put = PutObject.from_dict(event["data"]) except ValueError as err: @@ -75,6 +80,10 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: builder.add_put(put.kind, put.key, put.version, put.object) elif event["event"] == EventName.DELETE_OBJECT: + if has_unsupported_kind(event["data"]): + log.debug("Ignoring delete-object with unsupported kind: %s", event["data"].get("kind")) + continue + try: delete_object = DeleteObject.from_dict(event["data"]) except ValueError as err: diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index cb0428d8..903d5438 100644 --- a/ldclient/impl/datasourcev2/streaming_common.py +++ b/ldclient/impl/datasourcev2/streaming_common.py @@ -16,7 +16,8 @@ Error, EventName, Goodbye, - PutObject + PutObject, + has_unsupported_kind ) from ldclient.impl.util import ( _LD_ENVID_HEADER, @@ -65,12 +66,22 @@ def process_message( return None if msg.event == EventName.PUT_OBJECT: - put = PutObject.from_dict(json.loads(msg.data)) + data = json.loads(msg.data) + if has_unsupported_kind(data): + log.debug("Ignoring put-object with unsupported kind: %s", data.get("kind")) + return None + + put = PutObject.from_dict(data) change_set_builder.add_put(put.kind, put.key, put.version, put.object) return None if msg.event == EventName.DELETE_OBJECT: - delete = DeleteObject.from_dict(json.loads(msg.data)) + data = json.loads(msg.data) + if has_unsupported_kind(data): + log.debug("Ignoring delete-object with unsupported kind: %s", data.get("kind")) + return None + + delete = DeleteObject.from_dict(data) change_set_builder.add_delete(delete.kind, delete.key, delete.version) return None diff --git a/ldclient/impl/datasystem/protocolv2.py b/ldclient/impl/datasystem/protocolv2.py index 7741a216..ce2e1411 100644 --- a/ldclient/impl/datasystem/protocolv2.py +++ b/ldclient/impl/datasystem/protocolv2.py @@ -7,6 +7,18 @@ from ldclient.interfaces import EventName, ObjectKind +_SUPPORTED_KINDS = tuple(kind.value for kind in ObjectKind) + + +def has_unsupported_kind(data: dict) -> bool: + """ + Returns True if a put-object or delete-object payload names a kind this + SDK does not support. Such objects should be ignored so that new kinds can + be added to the protocol without breaking older SDKs. + """ + kind = data.get("kind") + return kind is not None and kind not in _SUPPORTED_KINDS + @dataclass(frozen=True) class DeleteObject: diff --git a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py index 8e4c6369..5f9d1676 100644 --- a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py +++ b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py @@ -110,6 +110,31 @@ def test_processes_delete_object(): assert change_set.selector.version == 461 +def test_ignores_objects_with_unknown_kinds(): + data = { + "events": [ + {"event": "server-intent", "data": {"payloads": [{"id": "p", "target": 1, "intentCode": "xfer-full", "reason": "payload-missing"}]}}, + {"event": "put-object", "data": {"version": 1, "kind": "some-future-kind", "key": "unknown-key", "object": {"foo": 1}}}, + {"event": "put-object", "data": {"version": 1, "kind": "flag", "key": "flag-key", "object": {"key": "flag-key", "version": 1}}}, + {"event": "delete-object", "data": {"version": 2, "kind": "some-future-kind", "key": "unknown-key"}}, + {"event": "delete-object", "data": {"version": 3, "kind": "segment", "key": "segment-key"}}, + {"event": "payload-transferred", "data": {"state": "(p:p:1)", "version": 1}}, + ] + } + # Parse a payload that mixes unknown kinds with known flag and segment changes. + result = polling_payload_to_changeset(data) + + # The payload completes with only the known changes. + assert isinstance(result, _Success) + + change_set = result.value + assert [(c.action, c.kind, c.key) for c in change_set.changes] == [ + (ChangeType.PUT, ObjectKind.FLAG, "flag-key"), + (ChangeType.DELETE, ObjectKind.SEGMENT, "segment-key"), + ] + assert change_set.selector.state == "(p:p:1)" + + def test_handles_invalid_put_object(): payload_str = '{"events":[ {"event":"server-intent","data":{"payloads":[ {"id":"5A46PZ79FQ9D08YYKT79DECDNV","target":461,"intentCode":"xfer-full","reason":"payload-missing"}]}},{"event": "put-object","data": {}},{"event":"payload-transferred","data":{"state":"(p:5A46PZ79FQ9D08YYKT79DECDNV:461)","id":"5A46PZ79FQ9D08YYKT79DECDNV","version":461}}]}' result = polling_payload_to_changeset(json.loads(payload_str)) diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 216013b4..1c1fab81 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -295,6 +295,44 @@ def test_handles_delete_objects(events): # pylint: disable=redefined-outer-name assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL +def test_ignores_objects_with_unknown_kinds(events): # pylint: disable=redefined-outer-name + unknown_put_event = Event( + event=EventName.PUT_OBJECT, + data=json.dumps({"version": 1, "kind": "some-future-kind", "key": "unknown-key", "object": {"foo": 1}}), + ) + unknown_delete_event = Event( + event=EventName.DELETE_OBJECT, + data=json.dumps({"version": 2, "kind": "some-future-kind", "key": "unknown-key"}), + ) + builder = list_sse_client( + [ + events[EventName.SERVER_INTENT], + unknown_put_event, + events[EventName.PUT_OBJECT], + unknown_delete_event, + events[EventName.DELETE_OBJECT], + events[EventName.PAYLOAD_TRANSFERRED], + ] + ) + + synchronizer = make_streaming_data_source() + synchronizer._sse_client_builder = builder + # Stream a payload that mixes unknown kinds with known flag changes. + updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + # The payload completes with only the known changes and no error. + assert len(updates) == 1 + assert updates[0].state == DataSourceState.VALID + assert updates[0].error is None + + assert updates[0].change_set is not None + assert [(c.action, c.kind, c.key) for c in updates[0].change_set.changes] == [ + (ChangeType.PUT, ObjectKind.FLAG, "flag-key"), + (ChangeType.DELETE, ObjectKind.FLAG, "flag-key"), + ] + assert updates[0].change_set.selector.state == "p:SOMETHING:300" + + def test_swallows_goodbye(events): # pylint: disable=redefined-outer-name builder = list_sse_client( [