GH-49697: [C++][CI] Check IPC file body bounds are in sync with decoder outcome

cpp/src/arrow/ipc/message.cc

@@ -423,10 +423,12 @@ static Result<std::unique_ptr<Message>> ReadMessageInternal(
             body, file->ReadAt(offset + metadata_length, decoder.next_required_size()));
       }
 
-      if (body->size() < decoder.next_required_size()) {
-        return Status::IOError("Expected to be able to read ",
-                               decoder.next_required_size(),
-                               " bytes for message body, got ", body->size());
+      if (body->size() != decoder.next_required_size()) {
+        // The streaming decoder got out of sync with the actual advertised
+        // metadata and body size, which signals an invalid IPC file.
+        return Status::IOError("Invalid IPC file: advertised body size is ", body->size(),
+                               ", but message decoder expects to read ",
+                               decoder.next_required_size(), " bytes instead");
       }
       RETURN_NOT_OK(decoder.Consume(body));
       return result;

cpp/src/arrow/ipc/reader.cc

@@ -624,6 +624,7 @@ Result<std::shared_ptr<RecordBatch>> LoadRecordBatchSubset(
   FieldVector filtered_fields;
   std::shared_ptr<Schema> filtered_schema;
 
+  // TODO factor this out?
   for (int i = 0; i < schema->num_fields(); ++i) {
     const Field& field = *schema->field(i);
     if (!inclusion_mask || (*inclusion_mask)[i]) {
@@ -645,6 +646,8 @@ Result<std::shared_ptr<RecordBatch>> LoadRecordBatchSubset(
     }
   }
 
+  // TODO factor out these steps?
+
   // Dictionary resolution needs to happen on the unfiltered columns,
   // because fields are mapped structurally (by path in the original schema).
   RETURN_NOT_OK(ResolveDictionaries(columns, *context.dictionary_memo,