fix: address possible desync framing scenarios

.coveragerc

@@ -6,7 +6,6 @@ omit =
 
 [report]
 show_missing = True
-skip_covered = True
 exclude_lines =
     pragma: no cover
     if __name__ == .__main__.

.gitignore

@@ -5,6 +5,12 @@
 .project
 .pydevproject
 .settings
+
+# Ignore JetBrains IDE Files
+.idea/
+
+# Ignore Claude Code Files
+.claude/
 *.sw*
 .coverage
 .pytest_cache/

README.md

@@ -132,7 +132,9 @@ Behavior-Driven Tests (Behave)
 Notes
 -----
 
-- Message framing uses a 4‑byte big‑endian length header followed by a JSON payload encoded as UTF‑8.
+- Breaking change: version 2.0.0 uses a new framing header (magic + length + CRC32). v1 clients are incompatible.
+- Message framing uses a 12‑byte header: 4‑byte magic, 4‑byte big‑endian length, and 4‑byte CRC32 of the payload, followed by a JSON payload encoded as UTF‑8.
+- `max_message_size` defaults to 10MB; set `.max_message_size` to adjust or set to `None` to disable.
 - On disconnect, reads raise `RuntimeError("socket connection broken")` so callers can distinguish cleanly from timeouts.
 - Binding with `port=0` lets the OS choose an ephemeral port; find it with `server.socket.getsockname()`.
 

docs/design-patterns.md

@@ -0,0 +1,62 @@
+# Design Patterns in JSocket
+
+This repo uses a few classic object-oriented design patterns. The diagrams below map directly to the current class structure and control flow.
+
+## Template Method (Message Processing Loop)
+
+`ThreadedServer` and `ServerFactoryThread` define the skeleton of a read-process-respond loop, and delegate the message-specific behavior to `_process_message`, which subclasses implement.
+
+```mermaid
+flowchart TD
+    A["ThreadedServer run"] --> B["handle client messages"]
+    B --> C["read obj"]
+    C --> D["process message"]
+    D --> E{response}
+    E -->|yes| F["send obj"]
+    E -->|no| G["continue loop"]
+```
+
+```mermaid
+classDiagram
+    class ThreadedServer {
+        +run()
+        +handle_client_messages()
+        #process_message(obj)
+    }
+    class EchoServer {
+        +process_message(obj)
+    }
+    ThreadedServer <|-- EchoServer
+```
+
+## Factory (Thread-per-Connection Workers)
+
+`ServerFactory` accepts a `ServerFactoryThread` subclass and instantiates a new worker per connection. This is effectively a factory for connection handlers.
+
+```mermaid
+classDiagram
+    class ServerFactory {
+        -thread_type
+        +run()
+    }
+    class ServerFactoryThread {
+        +swap_socket(sock)
+        +run()
+        #process_message(obj)
+    }
+    ServerFactory --> ServerFactoryThread : creates
+```
+
+## Facade (Simplified Public API)
+
+The top-level `jsocket` package re-exports the primary classes so callers can import from a single module instead of multiple internal modules.
+
+```mermaid
+flowchart LR
+    A["Application Code"] --> B["jsocket package"]
+    B --> C["jsocket_base JsonSocket"]
+    B --> D["jsocket_base JsonServer"]
+    B --> E["jsocket_base JsonClient"]
+    B --> F["tserver ThreadedServer"]
+    B --> G["tserver ServerFactory"]
+```

jsocket/_version.py

@@ -1,3 +1,3 @@
 """Package version."""
 
-__version__ = "1.9.6"
+__version__ = "2.0.0"

jsocket/jsocket_base.py

@@ -24,11 +24,21 @@
 import struct
 import logging
 import time
+import zlib
 
 from ._version import __version__
 
 logger = logging.getLogger("jsocket")
 
+FRAME_MAGIC = b"JSN1"
+FRAME_HEADER_FMT = "!4sII"
+FRAME_HEADER_SIZE = struct.calcsize(FRAME_HEADER_FMT)
+DEFAULT_MAX_MESSAGE_SIZE = 10 * 1024 * 1024
+
+
+class FramingError(RuntimeError):
+    """Raised when a message fails framing or integrity checks."""
+
 
 def _socket_fileno(sock):
     try:
@@ -40,12 +50,14 @@ def _socket_fileno(sock):
 class JsonSocket:
     """Lightweight JSON-over-TCP socket wrapper with length-prefixed framing."""
 
-    def __init__(self, address='127.0.0.1', port=5489, timeout=2.0):
+    def __init__(self, address='127.0.0.1', port=5489, timeout=2.0, max_message_size=DEFAULT_MAX_MESSAGE_SIZE):
         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self.conn = self.socket
         self._timeout = timeout
         self._address = address
         self._port = port
+        self._max_message_size = None
+        self.max_message_size = max_message_size
         self._last_client_addr = None
         # Ensure the primary socket respects timeout for accept/connect operations
         self.socket.settimeout(self._timeout)
@@ -55,41 +67,66 @@ def send_obj(self, obj):
         msg = json.dumps(obj, ensure_ascii=False)
         if self.socket:
             payload = msg.encode('utf-8')
-            frmt = f"={len(payload)}s"
-            packed_msg = struct.pack(frmt, payload)
-            packed_hdr = struct.pack('!I', len(packed_msg))
+            if self._max_message_size is not None and len(payload) > self._max_message_size:
+                raise ValueError(f"message exceeds max_message_size ({len(payload)} > {self._max_message_size})")
+            checksum = zlib.crc32(payload) & 0xFFFFFFFF
+            packed_hdr = struct.pack(FRAME_HEADER_FMT, FRAME_MAGIC, len(payload), checksum)
             self._send(packed_hdr)
-            self._send(packed_msg)
+            self._send(payload)
 
     def _send(self, msg):
         """Send all bytes in `msg` to the peer."""
         sent = 0
         while sent < len(msg):
             sent += self.conn.send(msg[sent:])
 
-    def _read(self, size):
+    def _read(self, size, allow_timeout=False):
         """Read exactly `size` bytes from the peer or raise on disconnect."""
         data = b''
         while len(data) < size:
-            data_tmp = self.conn.recv(size - len(data))
-            data += data_tmp
+            try:
+                data_tmp = self.conn.recv(size - len(data))
+            except socket.timeout:
+                if allow_timeout and not data:
+                    raise
+                self._close_connection()
+                raise FramingError("socket read timeout during message")
             if data_tmp == b'':
+                self._close_connection()
                 raise RuntimeError("socket connection broken")
+            data += data_tmp
         return data
 
-    def _msg_length(self):
-        """Read and unpack the 4-byte big-endian length header."""
-        d = self._read(4)
-        s = struct.unpack('!I', d)
-        return s[0]
+    def _read_header(self):
+        """Read and unpack the framing header."""
+        header = self._read(FRAME_HEADER_SIZE, allow_timeout=True)
+        magic, size, checksum = struct.unpack(FRAME_HEADER_FMT, header)
+        if magic != FRAME_MAGIC:
+            self._close_connection()
+            raise FramingError("invalid message header magic")
+        if self._max_message_size is not None and size > self._max_message_size:
+            self._close_connection()
+            raise FramingError(f"message length {size} exceeds max_message_size {self._max_message_size}")
+        return size, checksum
 
     def read_obj(self):
         """Read a full message and decode it as JSON, returning a Python object."""
-        size = self._msg_length()
+        size, checksum = self._read_header()
         data = self._read(size)
-        frmt = f"={size}s"
-        msg = struct.unpack(frmt, data)
-        return json.loads(msg[0].decode('utf-8'))
+        actual = zlib.crc32(data) & 0xFFFFFFFF
+        if actual != checksum:
+            self._close_connection()
+            raise FramingError("message checksum mismatch")
+        try:
+            decoded = data.decode('utf-8')
+        except UnicodeDecodeError as e:
+            self._close_connection()
+            raise FramingError("invalid UTF-8 payload") from e
+        try:
+            return json.loads(decoded)
+        except json.JSONDecodeError as e:
+            self._close_connection()
+            raise FramingError("invalid JSON payload") from e
 
     def close(self):
         """Close active connection and the listening socket if open."""
@@ -118,10 +155,10 @@ def _close_socket(self):
             pass
 
     def _close_connection(self):
-        """Best-effort shutdown and close of the accepted connection socket."""
+        """Best-effort shutdown and close of the connection socket."""
         logger.debug("closing connection socket (fd=%s)", _socket_fileno(self.conn))
         try:
-            if self.conn and self.conn is not self.socket and self.conn.fileno() != -1:
+            if self.conn and self.conn.fileno() != -1:
                 try:
                     self.conn.shutdown(socket.SHUT_RDWR)
                 except OSError:
@@ -158,9 +195,24 @@ def _set_port(self, _port):
         """No-op: port is read-only after initialization."""
         return None
 
+    def _get_max_message_size(self):
+        """Get the maximum allowed message size in bytes."""
+        return self._max_message_size
+
+    def _set_max_message_size(self, size):
+        """Set the maximum allowed message size in bytes."""
+        if size is None:
+            self._max_message_size = None
+            return
+        size = int(size)
+        if size <= 0:
+            raise ValueError("max_message_size must be positive")
+        self._max_message_size = size
+
     timeout = property(_get_timeout, _set_timeout, doc='Get/set the socket timeout')
     address = property(_get_address, _set_address, doc='read only property socket address')
     port = property(_get_port, _set_port, doc='read only property socket port')
+    max_message_size = property(_get_max_message_size, _set_max_message_size, doc='Get/set max message size in bytes')
 
 
 class JsonServer(JsonSocket):
@@ -170,6 +222,22 @@ def __init__(self, address='127.0.0.1', port=5489):
         super().__init__(address, port)
         self._bind()
 
+    def _close_connection(self):
+        """Best-effort shutdown and close of the accepted connection socket."""
+        logger.debug("closing connection socket (fd=%s)", _socket_fileno(self.conn))
+        try:
+            if self.conn and self.conn is not self.socket and self.conn.fileno() != -1:
+                try:
+                    self.conn.shutdown(socket.SHUT_RDWR)
+                except OSError:
+                    pass
+                try:
+                    self.conn.close()
+                except OSError:
+                    pass
+        except OSError:
+            pass
+
     def _bind(self):
         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         self.socket.bind((self.address, self.port))