diff --git a/Sources/GRPC/GRPCClientChannelHandler.swift b/Sources/GRPC/GRPCClientChannelHandler.swift
index ecc0c30a1..5f3c44034 100644
--- a/Sources/GRPC/GRPCClientChannelHandler.swift
+++ b/Sources/GRPC/GRPCClientChannelHandler.swift
@@ -463,6 +463,8 @@ extension GRPCClientChannelHandler: ChannelInboundHandler {
       case let .decompressionLimitExceeded(compressedSize):
         return GRPCError.DecompressionLimitExceeded(compressedSize: compressedSize)
           .captureContext()
+      case let .lengthExceedsLimit(underlyingError):
+        return underlyingError.captureContext()
       case .invalidState:
         return GRPCError.InvalidState("parsing data as a response message").captureContext()
       }
diff --git a/Sources/GRPC/GRPCError.swift b/Sources/GRPC/GRPCError.swift
index 14a6a5b6a..2692d4ecb 100644
--- a/Sources/GRPC/GRPCError.swift
+++ b/Sources/GRPC/GRPCError.swift
@@ -92,6 +92,19 @@ public enum GRPCError {
     }
   }
 
+  /// The length of the received payload was longer than is permitted.
+  public struct PayloadLengthLimitExceeded: GRPCErrorProtocol {
+    public let description: String
+
+    public init(actualLength length: Int, limit: Int) {
+      self.description = "Payload length exceeds limit (\(length) > \(limit))"
+    }
+
+    public func makeGRPCStatus() -> GRPCStatus {
+      return GRPCStatus(code: .resourceExhausted, message: self.description)
+    }
+  }
+
   /// It was not possible to compress or decompress a message with zlib.
   public struct ZlibCompressionFailure: GRPCErrorProtocol {
     var code: Int32
diff --git a/Sources/GRPC/LengthPrefixedMessageReader.swift b/Sources/GRPC/LengthPrefixedMessageReader.swift
index ffacc6885..c3f49c47e 100644
--- a/Sources/GRPC/LengthPrefixedMessageReader.swift
+++ b/Sources/GRPC/LengthPrefixedMessageReader.swift
@@ -185,7 +185,10 @@ internal struct LengthPrefixedMessageReader {
       }
 
       if messageLength > maxLength {
-        throw GRPCError.DeserializationFailure().captureContext()
+        throw GRPCError.PayloadLengthLimitExceeded(
+          actualLength: messageLength,
+          limit: maxLength
+        ).captureContext()
       }
 
       self.state = .expectingMessage(messageLength, compressed: compressed)
diff --git a/Sources/GRPC/ReadWriteStates.swift b/Sources/GRPC/ReadWriteStates.swift
index b865864fe..27df2d4c4 100644
--- a/Sources/GRPC/ReadWriteStates.swift
+++ b/Sources/GRPC/ReadWriteStates.swift
@@ -158,12 +158,15 @@ enum ReadState {
         }
       } catch {
         self = .notReading
-        if let grpcError = error as? GRPCError.WithContext,
-          let limitExceeded = grpcError.error as? GRPCError.DecompressionLimitExceeded {
-          return .failure(.decompressionLimitExceeded(limitExceeded.compressedSize))
-        } else {
-          return .failure(.deserializationFailed)
+        if let grpcError = error as? GRPCError.WithContext {
+          if let compressionLimit = grpcError.error as? GRPCError.DecompressionLimitExceeded {
+            return .failure(.decompressionLimitExceeded(compressionLimit.compressedSize))
+          } else if let lengthLimit = grpcError.error as? GRPCError.PayloadLengthLimitExceeded {
+            return .failure(.lengthExceedsLimit(lengthLimit))
+          }
         }
+
+        return .failure(.deserializationFailed)
       }
 
       // We need to validate the number of messages we decoded. Zero is fine because the payload may
@@ -209,6 +212,9 @@ enum MessageReadError: Error, Equatable {
   /// The limit for decompression was exceeded.
   case decompressionLimitExceeded(Int)
 
+  /// The length of the message exceeded the permitted maximum length.
+  case lengthExceedsLimit(GRPCError.PayloadLengthLimitExceeded)
+
   /// An invalid state was encountered. This is a serious implementation error.
   case invalidState
 }
diff --git a/Tests/GRPCTests/GRPCMessageLengthLimitTests.swift b/Tests/GRPCTests/GRPCMessageLengthLimitTests.swift
index 883478a61..b62406236 100644
--- a/Tests/GRPCTests/GRPCMessageLengthLimitTests.swift
+++ b/Tests/GRPCTests/GRPCMessageLengthLimitTests.swift
@@ -70,7 +70,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
 
     let get = self.echo.get(self.makeRequest(minimumLength: 1024))
     XCTAssertThrowsError(try get.response.wait())
-    XCTAssertEqual(try get.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try get.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testServerRejectsLongClientStreamingRequest() throws {
@@ -83,7 +83,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
     // (No need to send end, the server is going to close the RPC because the message was too long.)
 
     XCTAssertThrowsError(try collect.response.wait())
-    XCTAssertEqual(try collect.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try collect.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testServerRejectsLongServerStreamingRequest() throws {
@@ -94,7 +94,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
       XCTFail("Unexpected response")
     }
 
-    XCTAssertEqual(try expand.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try expand.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testServerRejectsLongBidirectionalStreamingRequest() throws {
@@ -107,7 +107,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
     XCTAssertNoThrow(try update.sendMessage(self.makeRequest(minimumLength: 1024)).wait())
     // (No need to send end, the server is going to close the RPC because the message was too long.)
 
-    XCTAssertEqual(try update.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try update.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testClientRejectsLongUnaryResponse() throws {
@@ -117,7 +117,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
 
     let get = self.echo.get(.with { $0.text = String(repeating: "x", count: 1024) })
     XCTAssertThrowsError(try get.response.wait())
-    XCTAssertEqual(try get.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try get.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testClientRejectsLongClientStreamingResponse() throws {
@@ -130,7 +130,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
     XCTAssertNoThrow(try collect.sendEnd().wait())
 
     XCTAssertThrowsError(try collect.response.wait())
-    XCTAssertEqual(try collect.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try collect.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testClientRejectsLongServerStreamingRequest() throws {
@@ -143,7 +143,7 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
       XCTFail("Unexpected response")
     }
 
-    XCTAssertEqual(try expand.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try expand.status.map { $0.code }.wait(), .resourceExhausted)
   }
 
   func testClientRejectsLongServerBidirectionalStreamingResponse() throws {
@@ -157,6 +157,6 @@ final class GRPCMessageLengthLimitTests: GRPCTestCase {
     // (No need to send end, the client will close the RPC when it receives a response which is too
     // long.
 
-    XCTAssertEqual(try update.status.map { $0.code }.wait(), .internalError)
+    XCTAssertEqual(try update.status.map { $0.code }.wait(), .resourceExhausted)
   }
 }