Port: Stream json garbage infinite loop and log json stream error

Source/core/JSON.h

@@ -449,7 +449,7 @@ namespace Core {
             const size_t nullTagLen = strlen(IElement::NullTag);
             ASSERT(offset < nullTagLen);
             while (offset < nullTagLen) {
-                if (loaded + 1 == maxLength) {
+                if (loaded >= maxLength) {
                     validity = ValueValidity::UNKNOWN;
                     break;
                 }

Source/core/StreamJSON.h

@@ -113,13 +113,15 @@ namespace Core {
                 , _factory(slotSize)
                 , _current()
                 , _offset(0)
+                , _consecutiveErrors(0)
             {
             }
             DeserializerImpl(ParentClass& parent, ALLOCATOR allocator)
                 : _parent(parent)
                 , _factory(allocator)
                 , _current()
                 , _offset(0)
+                , _consecutiveErrors(0)
             {
 			}
             ~DeserializerImpl()
@@ -140,29 +142,57 @@ namespace Core {
                     _offset = 0;
                 }
                 if (_current.IsValid() == true) {
-                    loaded = Deserialize(_current, stream, length);
-                    if ((_offset == 0) || (loaded != length)) {
-                        _parent.Received(_current);
-                        _current.Release();
+                    Core::OptionalType<Core::JSON::Error> error;
+
+                    loaded = Deserialize(_current, stream, length, error);
+
+                    if (error.IsSet() == true) {
+                        Clear(_current);
+                        _offset = 0;
+                        if (_consecutiveErrors < maxConsecutiveErrors) {
+                            ++_consecutiveErrors;
+
+                            error.Value().Context(reinterpret_cast<const char*>(stream), length, loaded);
+                            if (_consecutiveErrors < maxConsecutiveErrors) {
+                                CC_SYSLOG("StreamJSONType failed: %s", ErrorDisplayMessage(error.Value()).c_str());
+                            } else {
+                                CC_SYSLOG("StreamJSONType failed (errors after this one will not be reported until correct message is received again): %s", ErrorDisplayMessage(error.Value()).c_str());
+                            }
+                        }
+                    } else {
+                        _consecutiveErrors = 0;
+                        if ((_offset == 0) || (loaded != length)) {
+                            _parent.Received(_current);
+                            _current.Release();
+                        }
                     }
                 }
 
                 return (loaded);
             }
 
         private:
-            inline uint16_t Deserialize(const Core::ProxyType<Core::JSON::IElement>& source, const uint8_t* stream, const uint16_t length) {
-                return(source->Deserialize(reinterpret_cast<const char*>(stream), length, _offset));
+            inline uint16_t Deserialize(const Core::ProxyType<Core::JSON::IElement>& source, const uint8_t* stream, const uint16_t length, Core::OptionalType<Core::JSON::Error>& error) {
+                return(source->Deserialize(reinterpret_cast<const char*>(stream), length, _offset, error));
             }
-            inline uint16_t Deserialize(const Core::ProxyType<Core::JSON::IMessagePack>& source, const uint8_t* stream, const uint16_t length) {
+            inline uint16_t Deserialize(const Core::ProxyType<Core::JSON::IMessagePack>& source, const uint8_t* stream, const uint16_t length, Core::OptionalType<Core::JSON::Error>&) {
                 return (source->Deserialize(stream, length, _offset));
             }
+            inline void Clear(const Core::ProxyType<Core::JSON::IElement>& source) {
+                source->Clear();
+            }
+            inline void Clear(const Core::ProxyType<Core::JSON::IMessagePack>& source) {
+                source->Clear();
+            }
+        private:
+            static constexpr uint8_t maxConsecutiveErrors = 200;
 
         private:
             ParentClass& _parent;
             ALLOCATOR _factory;
             Core::ProxyType<INTERFACE> _current;
             uint32_t _offset;
+            uint8_t _consecutiveErrors;
         };
 
         class HandlerType : public SOURCE {
@@ -287,12 +317,14 @@ POP_WARNING()
         uint16_t ReceiveData(uint8_t* dataFrame, const uint16_t receivedSize)
         {
             uint16_t handled = 0;
+            uint16_t processed;
 
             do {
-                handled += _deserializer.Deserialize(&dataFrame[handled], (receivedSize - handled));
+                processed = _deserializer.Deserialize(&dataFrame[handled], (receivedSize - handled));
+                handled += processed;
 
                 // The dataframe can hold more items....
-            } while (handled < receivedSize);
+            } while ((processed != 0) && (handled < receivedSize));
 
             return (handled);
         }

Tests/CMakeLists.txt

@@ -2,6 +2,7 @@ option(LOADER_TEST "Utility to load a plugin in isolation." OFF)
 option(HTTPSCLIENT_TEST "Example how to do https requests with Thunder." OFF)
 option(WORKERPOOL_TEST "WorkerPool stress test" OFF)
 option(FILE_UNLINK_TEST "File unlink test" OFF)
+option(STREAMJSON_GARBAGE_TEST "Reproducer for issue #1963: infinite loop on garbage data in StreamJSONType::ReceiveData()" OFF)
 
 if(BUILD_TESTS)
     add_subdirectory(unit)
@@ -22,3 +23,7 @@ endif()
 if(WORKERPOOL_TEST)
     add_subdirectory(workerpool-test)
 endif()
+
+if(STREAMJSON_GARBAGE_TEST)
+    add_subdirectory(streamjson-garbage)
+endif()

Tests/streamjson-garbage/CMakeLists.txt

@@ -0,0 +1,24 @@
+if(NOT CMAKE_SYSTEM_NAME STREQUAL "Windows")
+add_executable(StreamJSONGarbageTest
+    Module.cpp
+    GarbageTest.cpp
+)
+
+set_target_properties(StreamJSONGarbageTest PROPERTIES
+    CXX_STANDARD ${CXX_STD}
+    CXX_STANDARD_REQUIRED YES
+)
+
+target_compile_options(StreamJSONGarbageTest PRIVATE -pthread)
+target_link_options(StreamJSONGarbageTest PRIVATE -pthread)
+
+target_link_libraries(StreamJSONGarbageTest
+    PRIVATE
+        ${NAMESPACE}Core
+)
+
+install(TARGETS StreamJSONGarbageTest
+    DESTINATION ${CMAKE_INSTALL_BINDIR}
+    COMPONENT ${NAMESPACE}_Test
+)
+endif() # NOT Windows