GH-48844: [C++] Check IPC Message body length consistency in IPC file (#48845)

### Rationale for this change

The IPC file exposes [redundant information](https://github.com/apache/arrow/blob/d54a2051cf9020a0fdf50836420c38ad14787abb/format/File.fbs#L39-L50) about Message sizes so as to allow for random access from the file footer.

We tried adding [consistency checks](#19596) in the past but this hit a bug in the JavaScript IPC writer at the time, so the checks were left disabled.

The JavaScript implementation was fixed soon after (7 years ago), so this PR re-enables those checks so as to more easily detect potentially invalid IPC files.

### Are these changes tested?

By existing tests.

### Are there any user-facing changes?

No, unless they try reading invalid IPC files.

* GitHub Issue: #48844

Authored-by: Antoine Pitrou <antoine@python.org>
Signed-off-by: Antoine Pitrou <antoine@python.org>

Author: pitrou

cpp/src/arrow/ipc/message.cc

@@ -375,6 +375,8 @@ Result<std::unique_ptr<Message>> ReadMessage(int64_t offset, int32_t metadata_le
                            decoder.next_required_size());
   }
 
+  // TODO(GH-48846): we should take a body_length just like ReadMessageAsync
+  // and read metadata + body in one go.
   ARROW_ASSIGN_OR_RAISE(auto metadata, file->ReadAt(offset, metadata_length));
   if (metadata->size() < metadata_length) {
     return Status::Invalid("Expected to read ", metadata_length,

cpp/src/arrow/ipc/reader.cc

@@ -1180,31 +1180,36 @@ Status CheckAligned(const FileBlock& block) {
   return Status::OK();
 }
 
+template <typename MessagePtr>
+Result<MessagePtr> CheckBodyLength(MessagePtr message, const FileBlock& block) {
+  if (message->body_length() != block.body_length) {
+    return Status::Invalid(
+        "Mismatching body length for IPC message "
+        "(Block.bodyLength: ",
+        block.body_length, " vs. Message.bodyLength: ", message->body_length(), ")");
+  }
+  // NOTE: we cannot check metadata length as easily as we would have to account
+  // for the additional IPC signalisation (such as optional continuation bytes).
+  return message;
+}
+
 Result<std::unique_ptr<Message>> ReadMessageFromBlock(
     const FileBlock& block, io::RandomAccessFile* file,
     const FieldsLoaderFunction& fields_loader) {
   RETURN_NOT_OK(CheckAligned(block));
-  // TODO(wesm): this breaks integration tests, see ARROW-3256
-  // DCHECK_EQ((*out)->body_length(), block.body_length);
-
   ARROW_ASSIGN_OR_RAISE(auto message, ReadMessage(block.offset, block.metadata_length,
                                                   file, fields_loader));
-  return message;
+  return CheckBodyLength(std::move(message), block);
 }
 
 Future<std::shared_ptr<Message>> ReadMessageFromBlockAsync(
     const FileBlock& block, io::RandomAccessFile* file, const io::IOContext& io_context) {
-  if (!bit_util::IsMultipleOf8(block.offset) ||
-      !bit_util::IsMultipleOf8(block.metadata_length) ||
-      !bit_util::IsMultipleOf8(block.body_length)) {
-    return Status::Invalid("Unaligned block in IPC file");
-  }
-
-  // TODO(wesm): this breaks integration tests, see ARROW-3256
-  // DCHECK_EQ((*out)->body_length(), block.body_length);
-
+  RETURN_NOT_OK(CheckAligned(block));
   return ReadMessageAsync(block.offset, block.metadata_length, block.body_length, file,
-                          io_context);
+                          io_context)
+      .Then([block](std::shared_ptr<Message> message) {
+        return CheckBodyLength(std::move(message), block);
+      });
 }
 
 class RecordBatchFileReaderImpl;