diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1e06bafa91..83eeeb31ea 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -24,6 +24,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](https://semver.org/).
 
+## [Unreleased]
+
+### Breaking Changes
+
+* `FileWrite::close`, `EncryptedOutputFile::write`, `ManifestListWriter::close`, and `PuffinWriter::close` now return `Result<io::FileMetadata>` instead of `Result<()>`. The returned size is the on-disk size, including encryption overhead. Custom `FileWrite` implementations must return the number of bytes stored.
+* `EncryptedOutputFile::key_metadata()` is replaced by `key_metadata_with_saved_file_metadata(&FileMetadata)`. Pass the metadata returned by `write()` or the writer's `close()` to include the stored length before encoding key metadata.
+* `EncryptedInputFile::metadata()` is now synchronous and derives the plaintext size from key metadata without a storage stat. Remove `.await` from calls to this method.
+* AGS1 readers now require `StandardKeyMetadata::file_length` and reject missing or invalid lengths without falling back to a storage stat. AGS1-encrypted manifests, manifest lists, and Puffin files written by earlier development builds without this field must be rewritten using a build that can still read them before upgrading. This matches the Java client's read contract.
+
 ## [v0.10.1] - 2026-07-28
 
 * chore: Update crossbeam-epoch to 0.9.20, pin CI ruff version to 0.15.22 by @dannycjones in https://github.com/apache/iceberg-rust/pull/2911
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index a917d28f5a..8398b3cdfb 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -273,7 +273,7 @@ pub struct iceberg::encryption::AesGcmFileWrite
 impl iceberg::encryption::AesGcmFileWrite
 pub fn iceberg::encryption::AesGcmFileWrite::new(inner: alloc::boxed::Box<dyn iceberg::io::FileWrite>, cipher: alloc::sync::Arc<iceberg::encryption::AesGcmCipher>, aad_prefix: impl core::convert::Into<alloc::boxed::Box<[u8]>>) -> Self
 impl iceberg::io::FileWrite for iceberg::encryption::AesGcmFileWrite
-pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
+pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub fn iceberg::encryption::AesGcmFileWrite::write<'life0, 'async_trait>(&'life0 mut self, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub struct iceberg::encryption::EncryptedInputFile
 impl iceberg::encryption::EncryptedInputFile
@@ -281,7 +281,7 @@ pub async fn iceberg::encryption::EncryptedInputFile::exists(&self) -> iceberg::
 pub fn iceberg::encryption::EncryptedInputFile::into_inner(self) -> iceberg::io::InputFile
 pub fn iceberg::encryption::EncryptedInputFile::key_metadata(&self) -> &iceberg::encryption::StandardKeyMetadata
 pub fn iceberg::encryption::EncryptedInputFile::location(&self) -> &str
-pub async fn iceberg::encryption::EncryptedInputFile::metadata(&self) -> iceberg::Result<iceberg::io::FileMetadata>
+pub fn iceberg::encryption::EncryptedInputFile::metadata(&self) -> iceberg::Result<iceberg::io::FileMetadata>
 pub fn iceberg::encryption::EncryptedInputFile::new(inner: iceberg::io::InputFile, key_metadata: iceberg::encryption::StandardKeyMetadata) -> Self
 pub async fn iceberg::encryption::EncryptedInputFile::read(&self) -> iceberg::Result<bytes::bytes::Bytes>
 pub async fn iceberg::encryption::EncryptedInputFile::reader(&self) -> iceberg::Result<alloc::boxed::Box<dyn iceberg::io::FileRead>>
@@ -291,10 +291,10 @@ pub struct iceberg::encryption::EncryptedOutputFile
 impl iceberg::encryption::EncryptedOutputFile
 pub async fn iceberg::encryption::EncryptedOutputFile::delete(&self) -> iceberg::Result<()>
 pub fn iceberg::encryption::EncryptedOutputFile::into_inner(self) -> iceberg::io::OutputFile
-pub fn iceberg::encryption::EncryptedOutputFile::key_metadata(&self) -> &iceberg::encryption::StandardKeyMetadata
+pub fn iceberg::encryption::EncryptedOutputFile::key_metadata_with_saved_file_metadata(&self, file_metadata: &iceberg::io::FileMetadata) -> iceberg::encryption::StandardKeyMetadata
 pub fn iceberg::encryption::EncryptedOutputFile::location(&self) -> &str
 pub fn iceberg::encryption::EncryptedOutputFile::new(inner: iceberg::io::OutputFile, key_metadata: iceberg::encryption::StandardKeyMetadata) -> Self
-pub async fn iceberg::encryption::EncryptedOutputFile::write(&self, bs: bytes::bytes::Bytes) -> iceberg::Result<()>
+pub async fn iceberg::encryption::EncryptedOutputFile::write(&self, bs: bytes::bytes::Bytes) -> iceberg::Result<iceberg::io::FileMetadata>
 pub async fn iceberg::encryption::EncryptedOutputFile::writer(&self) -> iceberg::Result<alloc::boxed::Box<dyn iceberg::io::FileWrite>>
 impl core::fmt::Debug for iceberg::encryption::EncryptedOutputFile
 pub fn iceberg::encryption::EncryptedOutputFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
@@ -1046,10 +1046,10 @@ pub fn iceberg::encryption::AesGcmFileRead::read<'life0, 'async_trait>(&'life0 s
 impl<T: core::convert::AsRef<dyn iceberg::io::FileRead> + core::marker::Send + core::marker::Sync + core::marker::Unpin + 'static> iceberg::io::FileRead for T
 pub fn T::read<'life0, 'async_trait>(&'life0 self, range: core::ops::range::Range<u64>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<bytes::bytes::Bytes>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub trait iceberg::io::FileWrite: core::marker::Send + core::marker::Unpin + 'static
-pub fn iceberg::io::FileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
+pub fn iceberg::io::FileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub fn iceberg::io::FileWrite::write<'life0, 'async_trait>(&'life0 mut self, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 impl iceberg::io::FileWrite for iceberg::encryption::AesGcmFileWrite
-pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
+pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub fn iceberg::encryption::AesGcmFileWrite::write<'life0, 'async_trait>(&'life0 mut self, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub trait iceberg::io::Storage: core::fmt::Debug + core::marker::Send + core::marker::Sync + typetag::Serialize + typetag::Deserialize
 pub fn iceberg::io::Storage::delete<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
@@ -1267,7 +1267,7 @@ pub async fn iceberg::puffin::PuffinReader::new_from_encrypted(encrypted_input:
 pub struct iceberg::puffin::PuffinWriter
 impl iceberg::puffin::PuffinWriter
 pub async fn iceberg::puffin::PuffinWriter::add(&mut self, blob: iceberg::puffin::Blob, compression_codec: iceberg::compression::CompressionCodec) -> iceberg::Result<()>
-pub async fn iceberg::puffin::PuffinWriter::close(self) -> iceberg::Result<()>
+pub async fn iceberg::puffin::PuffinWriter::close(self) -> iceberg::Result<iceberg::io::FileMetadata>
 pub async fn iceberg::puffin::PuffinWriter::new(output_file: &iceberg::io::OutputFile, properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, compress_footer: bool) -> iceberg::Result<Self>
 pub async fn iceberg::puffin::PuffinWriter::new_from_encrypted(encrypted_output: &iceberg::encryption::EncryptedOutputFile, properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, compress_footer: bool) -> iceberg::Result<Self>
 pub const iceberg::puffin::APACHE_DATASKETCHES_THETA_V1: &str
@@ -2138,7 +2138,7 @@ pub async fn iceberg::spec::ManifestListReader::load(&self) -> iceberg::Result<i
 pub struct iceberg::spec::ManifestListWriter
 impl iceberg::spec::ManifestListWriter
 pub fn iceberg::spec::ManifestListWriter::add_manifests(&mut self, manifests: impl core::iter::traits::iterator::Iterator<Item = iceberg::spec::ManifestFile>) -> iceberg::Result<()>
-pub async fn iceberg::spec::ManifestListWriter::close(self) -> iceberg::Result<()>
+pub async fn iceberg::spec::ManifestListWriter::close(self) -> iceberg::Result<iceberg::io::FileMetadata>
 pub fn iceberg::spec::ManifestListWriter::next_row_id(&self) -> core::option::Option<u64>
 pub fn iceberg::spec::ManifestListWriter::v1(writer: alloc::boxed::Box<dyn iceberg::io::FileWrite>, snapshot_id: i64, parent_snapshot_id: core::option::Option<i64>) -> Self
 pub fn iceberg::spec::ManifestListWriter::v2(writer: alloc::boxed::Box<dyn iceberg::io::FileWrite>, snapshot_id: i64, parent_snapshot_id: core::option::Option<i64>, sequence_number: i64) -> Self
diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
index 905fa86d83..5b12491e66 100644
--- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
@@ -1647,13 +1647,16 @@ mod tests {
         let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
             .unwrap()
             .with_aad_prefix(b"test-aad-prefix!");
-        let encoded_key_metadata = key_metadata.encode().unwrap();
 
         let blob = encode_dv_blob([2u64, 4]);
         let plaintext_size = blob.len() as i64;
         let dv_path = format!("{table_location}/deletes.puffin");
         let output = EncryptedOutputFile::new(file_io.new_output(&dv_path).unwrap(), key_metadata);
-        output.write(Bytes::from(blob)).await.unwrap();
+        let file_metadata = output.write(Bytes::from(blob)).await.unwrap();
+        let encoded_key_metadata = output
+            .key_metadata_with_saved_file_metadata(&file_metadata)
+            .encode()
+            .unwrap();
 
         // content_offset / content_size_in_bytes are in the plaintext coordinate space, distinct
         // from the ciphertext's on-disk size (header, nonce, and tag overhead).
diff --git a/crates/iceberg/src/encryption/io.rs b/crates/iceberg/src/encryption/io.rs
index 3eafbe82f3..1169979668 100644
--- a/crates/iceberg/src/encryption/io.rs
+++ b/crates/iceberg/src/encryption/io.rs
@@ -23,9 +23,9 @@ use bytes::Bytes;
 
 use super::crypto::AesGcmCipher;
 use super::key_metadata::StandardKeyMetadata;
-use super::stream::{AesGcmFileRead, AesGcmFileWrite};
-use crate::Result;
+use super::stream::{AesGcmFileRead, AesGcmFileWrite, MIN_STREAM_LENGTH};
 use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile};
+use crate::{Error, ErrorKind, Result};
 
 /// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`].
 ///
@@ -54,12 +54,11 @@ impl EncryptedInputFile {
         self.inner.exists().await
     }
 
-    /// Fetch and returns metadata of file.
+    /// Returns file metadata from the declared encrypted length without performing I/O.
     ///
     /// The returned size is the **plaintext** size.
-    pub async fn metadata(&self) -> Result<FileMetadata> {
-        let raw_meta = self.inner.metadata().await?;
-        let plaintext_size = AesGcmFileRead::calculate_plaintext_length(raw_meta.size)?;
+    pub fn metadata(&self) -> Result<FileMetadata> {
+        let plaintext_size = AesGcmFileRead::calculate_plaintext_length(self.encrypted_length()?)?;
         Ok(FileMetadata {
             size: plaintext_size,
         })
@@ -67,21 +66,38 @@ impl EncryptedInputFile {
 
     /// Read and returns whole content of file (decrypted plaintext).
     pub async fn read(&self) -> Result<Bytes> {
-        let meta = self.metadata().await?;
+        let meta = self.metadata()?;
         let reader = self.reader().await?;
         reader.read(0..meta.size).await
     }
 
     /// Creates a reader that transparently decrypts on each read.
     pub async fn reader(&self) -> Result<Box<dyn FileRead>> {
-        let raw_meta = self.inner.metadata().await?;
+        let encrypted_length = self.encrypted_length()?;
         let raw_reader = self.inner.reader().await?;
         let cipher = build_cipher(&self.key_metadata)?;
         let aad_prefix: Box<[u8]> = self.key_metadata.aad_prefix().unwrap_or_default().into();
-        let decrypting = AesGcmFileRead::new(raw_reader, cipher, aad_prefix, raw_meta.size)?;
+        let decrypting = AesGcmFileRead::new(raw_reader, cipher, aad_prefix, encrypted_length)?;
         Ok(Box::new(decrypting))
     }
 
+    // A storage stat would hide truncation; require the original length from key metadata.
+    fn encrypted_length(&self) -> Result<u64> {
+        let length = self.key_metadata.file_length().ok_or_else(|| {
+            Error::new(
+                ErrorKind::DataInvalid,
+                "AGS1 key metadata is missing the encrypted file length",
+            )
+        })?;
+        if length < u64::from(MIN_STREAM_LENGTH) {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!("Invalid encrypted file length: {length} is less than {MIN_STREAM_LENGTH}"),
+            ));
+        }
+        Ok(length)
+    }
+
     /// Returns a reference to the file's key metadata.
     pub fn key_metadata(&self) -> &StandardKeyMetadata {
         &self.key_metadata
@@ -118,9 +134,14 @@ impl EncryptedOutputFile {
         }
     }
 
-    /// Returns a reference to the file's key metadata.
-    pub fn key_metadata(&self) -> &StandardKeyMetadata {
-        &self.key_metadata
+    /// Returns key metadata using the encrypted size returned by [`FileWrite::close`] or [`Self::write`].
+    pub fn key_metadata_with_saved_file_metadata(
+        &self,
+        file_metadata: &FileMetadata,
+    ) -> StandardKeyMetadata {
+        self.key_metadata
+            .clone()
+            .with_file_length(file_metadata.size)
     }
 
     /// Absolute path of the file.
@@ -138,8 +159,8 @@ impl EncryptedOutputFile {
         )))
     }
 
-    /// Write bytes to file (transparently encrypted).
-    pub async fn write(&self, bs: Bytes) -> Result<()> {
+    /// Write bytes to the file and return its encrypted size.
+    pub async fn write(&self, bs: Bytes) -> Result<FileMetadata> {
         let mut writer = self.writer().await?;
         writer.write(bs).await?;
         writer.close().await
@@ -172,6 +193,9 @@ fn build_cipher(metadata: &StandardKeyMetadata) -> Result<Arc<AesGcmCipher>> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::encryption::stream::{
+        CIPHER_BLOCK_SIZE, GCM_STREAM_HEADER_LENGTH, PLAIN_BLOCK_SIZE,
+    };
     use crate::io::FileIO;
 
     fn key_metadata() -> StandardKeyMetadata {
@@ -187,9 +211,12 @@ mod tests {
         let plaintext = b"Hello from EncryptedInputFile/EncryptedOutputFile!";
 
         let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
-        output.write(Bytes::from(plaintext.to_vec())).await.unwrap();
+        let file_metadata = output.write(Bytes::from(plaintext.to_vec())).await.unwrap();
 
-        let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), key_metadata());
+        let input = EncryptedInputFile::new(
+            fileio.new_input(path).unwrap(),
+            output.key_metadata_with_saved_file_metadata(&file_metadata),
+        );
         let content = input.read().await.unwrap();
         assert_eq!(&content[..], plaintext);
     }
@@ -201,7 +228,7 @@ mod tests {
         let plaintext = b"some bytes to measure";
 
         let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
-        output.write(Bytes::from(plaintext.to_vec())).await.unwrap();
+        let file_metadata = output.write(Bytes::from(plaintext.to_vec())).await.unwrap();
 
         let raw_size = fileio
             .new_input(path)
@@ -215,8 +242,172 @@ mod tests {
             "encrypted file should be larger than plaintext (header + nonce + tag)"
         );
 
-        let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), key_metadata());
-        let meta = input.metadata().await.unwrap();
+        // A missing path proves the size comes from the key metadata rather than a stat call.
+        let input = EncryptedInputFile::new(
+            fileio.new_input("memory:///does-not-exist").unwrap(),
+            output.key_metadata_with_saved_file_metadata(&file_metadata),
+        );
+        let meta = input.metadata().unwrap();
         assert_eq!(meta.size, plaintext.len() as u64);
     }
+
+    #[tokio::test]
+    async fn test_missing_file_length_is_rejected() {
+        let fileio = FileIO::new_with_memory();
+        let path = "memory:///test/missing_length.bin";
+        let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
+        output.write(Bytes::from_static(b"data")).await.unwrap();
+        let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), key_metadata());
+
+        for err in [
+            input.metadata().err().unwrap(),
+            input.reader().await.err().unwrap(),
+            input.read().await.unwrap_err(),
+        ] {
+            assert_eq!(err.kind(), ErrorKind::DataInvalid);
+            assert!(
+                err.to_string()
+                    .contains("missing the encrypted file length")
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_invalid_file_length_is_rejected() {
+        let fileio = FileIO::new_with_memory();
+        for length in [
+            0,
+            u64::from(GCM_STREAM_HEADER_LENGTH),
+            u64::from(MIN_STREAM_LENGTH - 1),
+        ] {
+            let input = EncryptedInputFile::new(
+                fileio
+                    .new_input("memory:///test/invalid_length.bin")
+                    .unwrap(),
+                key_metadata().with_file_length(length),
+            );
+            assert_eq!(
+                input.metadata().err().unwrap().kind(),
+                ErrorKind::DataInvalid
+            );
+            assert_eq!(
+                input.reader().await.err().unwrap().kind(),
+                ErrorKind::DataInvalid
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_oversized_file_length_is_rejected() {
+        let fileio = FileIO::new_with_memory();
+        let path = "memory:///test/oversized_length.bin";
+        let plaintext = Bytes::from_static(b"some bytes to measure");
+        let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
+        let file_metadata = output.write(plaintext.clone()).await.unwrap();
+
+        // A declared length is trusted without a stat, so an inflated one is only caught once a
+        // read runs off the end of the real file. Both a minimal overstatement and one spanning a
+        // whole extra block must fail rather than silently return short plaintext.
+        for excess in [1, u64::from(CIPHER_BLOCK_SIZE)] {
+            let input = EncryptedInputFile::new(
+                fileio.new_input(path).unwrap(),
+                key_metadata().with_file_length(file_metadata.size + excess),
+            );
+
+            let inflated_size = input.metadata().unwrap().size;
+            assert!(inflated_size > plaintext.len() as u64);
+
+            assert_eq!(
+                input.read().await.unwrap_err().kind(),
+                ErrorKind::DataInvalid
+            );
+
+            // Not even the bytes that genuinely are on disk can be read back.
+            let reader = input.reader().await.unwrap();
+            assert_eq!(
+                reader
+                    .read(0..plaintext.len() as u64)
+                    .await
+                    .unwrap_err()
+                    .kind(),
+                ErrorKind::DataInvalid
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_truncated_file_is_rejected() {
+        let fileio = FileIO::new_with_memory();
+        let path = "memory:///test/truncated.bin";
+        let plaintext = Bytes::from(vec![42; 2 * PLAIN_BLOCK_SIZE as usize + 17]);
+        let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
+        let file_metadata = output.write(plaintext.clone()).await.unwrap();
+        let metadata = key_metadata().with_file_length(file_metadata.size);
+        let ciphertext = fileio.new_input(path).unwrap().read().await.unwrap();
+        let truncated_length = (GCM_STREAM_HEADER_LENGTH + CIPHER_BLOCK_SIZE) as usize;
+        fileio
+            .new_output(path)
+            .unwrap()
+            .write(ciphertext.slice(..truncated_length))
+            .await
+            .unwrap();
+
+        let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), metadata);
+        assert_eq!(input.metadata().unwrap().size, plaintext.len() as u64);
+        let reader = input.reader().await.unwrap();
+        assert_eq!(
+            reader.read(0..u64::from(PLAIN_BLOCK_SIZE)).await.unwrap(),
+            plaintext.slice(..PLAIN_BLOCK_SIZE as usize)
+        );
+        assert_eq!(
+            input.read().await.unwrap_err().kind(),
+            ErrorKind::DataInvalid
+        );
+    }
+
+    #[tokio::test]
+    async fn test_truncated_empty_file_is_rejected() {
+        let fileio = FileIO::new_with_memory();
+        let path = "memory:///test/truncated_empty.bin";
+        let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
+        let file_metadata = output.write(Bytes::new()).await.unwrap();
+        let metadata = key_metadata().with_file_length(file_metadata.size);
+        let ciphertext = fileio.new_input(path).unwrap().read().await.unwrap();
+        fileio
+            .new_output(path)
+            .unwrap()
+            .write(ciphertext.slice(..GCM_STREAM_HEADER_LENGTH as usize))
+            .await
+            .unwrap();
+        let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), metadata);
+        assert_eq!(
+            input.read().await.unwrap_err().kind(),
+            ErrorKind::DataInvalid
+        );
+    }
+
+    #[tokio::test]
+    async fn test_close_returns_encrypted_size() {
+        let fileio = FileIO::new_with_memory();
+        let path = "memory:///test/streaming.bin";
+        let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
+        for plaintext in [
+            Bytes::from(vec![42; PLAIN_BLOCK_SIZE as usize + 17]),
+            Bytes::new(),
+        ] {
+            let mut writer = output.writer().await.unwrap();
+            for chunk in plaintext.chunks(1024) {
+                writer.write(Bytes::copy_from_slice(chunk)).await.unwrap();
+            }
+            let metadata = writer.close().await.unwrap();
+            let size = fileio
+                .new_input(path)
+                .unwrap()
+                .metadata()
+                .await
+                .unwrap()
+                .size;
+            assert_eq!(metadata.size, size);
+        }
+    }
 }
diff --git a/crates/iceberg/src/encryption/key_metadata.rs b/crates/iceberg/src/encryption/key_metadata.rs
index 765d600b50..7bc979c43e 100644
--- a/crates/iceberg/src/encryption/key_metadata.rs
+++ b/crates/iceberg/src/encryption/key_metadata.rs
@@ -74,7 +74,7 @@ impl StandardKeyMetadata {
         self
     }
 
-    /// Adds a file length.
+    /// Sets the encrypted file length in bytes, required for AGS1 truncation protection.
     pub fn with_file_length(mut self, length: u64) -> Self {
         self.file_length = Some(length);
         self
@@ -90,7 +90,7 @@ impl StandardKeyMetadata {
         self.aad_prefix.as_deref()
     }
 
-    /// Returns the optional file length.
+    /// Returns the optional encrypted file length in bytes; AGS1 readers require it to be set.
     pub fn file_length(&self) -> Option<u64> {
         self.file_length
     }
diff --git a/crates/iceberg/src/encryption/manager.rs b/crates/iceberg/src/encryption/manager.rs
index f54ef464d5..5f26d080da 100644
--- a/crates/iceberg/src/encryption/manager.rs
+++ b/crates/iceberg/src/encryption/manager.rs
@@ -688,12 +688,16 @@ mod tests {
         let encrypted_output = mgr.encrypt(output);
 
         let plaintext = b"Hello, encrypted Iceberg round-trip!";
-        let serialized_metadata = encrypted_output.key_metadata().encode().unwrap();
-        encrypted_output
+        let file_metadata = encrypted_output
             .write(bytes::Bytes::from(plaintext.to_vec()))
             .await
             .unwrap();
 
+        let serialized_metadata = encrypted_output
+            .key_metadata_with_saved_file_metadata(&file_metadata)
+            .encode()
+            .unwrap();
+
         let input = io.new_input(path).unwrap();
         let parsed_metadata = StandardKeyMetadata::decode(&serialized_metadata).unwrap();
         let decrypted_file = EncryptedInputFile::new(input, parsed_metadata);
diff --git a/crates/iceberg/src/encryption/stream.rs b/crates/iceberg/src/encryption/stream.rs
index 7971a3df54..745914cb01 100644
--- a/crates/iceberg/src/encryption/stream.rs
+++ b/crates/iceberg/src/encryption/stream.rs
@@ -49,7 +49,7 @@ use std::sync::Arc;
 use bytes::{Bytes, BytesMut};
 
 use super::AesGcmCipher;
-use crate::io::{FileRead, FileWrite};
+use crate::io::{FileMetadata, FileRead, FileWrite};
 use crate::{Error, ErrorKind, Result};
 
 /// Default plaintext block size (1 MiB), matching Java's `Ciphers.PLAIN_BLOCK_SIZE`.
@@ -71,8 +71,7 @@ pub const GCM_STREAM_MAGIC: [u8; 4] = *b"AGS1";
 pub const GCM_STREAM_HEADER_LENGTH: u32 = 8;
 
 /// Minimum valid AGS1 stream length (header + one empty block).
-#[cfg(test)]
-pub const MIN_STREAM_LENGTH: u32 = GCM_STREAM_HEADER_LENGTH + NONCE_LENGTH + GCM_TAG_LENGTH;
+pub(crate) const MIN_STREAM_LENGTH: u32 = GCM_STREAM_HEADER_LENGTH + NONCE_LENGTH + GCM_TAG_LENGTH;
 
 /// Constructs the per-block AAD for AGS1 stream encryption.
 ///
@@ -146,20 +145,17 @@ impl AesGcmFileRead {
         aad_prefix: Box<[u8]>,
         encrypted_file_length: u64,
     ) -> Result<Self> {
+        if encrypted_file_length < u64::from(MIN_STREAM_LENGTH) {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "Invalid encrypted file length: {encrypted_file_length} is less than {MIN_STREAM_LENGTH}"
+                ),
+            ));
+        }
         let plain_stream_size = Self::calculate_plaintext_length(encrypted_file_length)?;
         let stream_length = encrypted_file_length - GCM_STREAM_HEADER_LENGTH as u64;
 
-        if stream_length == 0 {
-            return Ok(Self {
-                inner,
-                cipher,
-                aad_prefix,
-                plain_stream_size: 0,
-                num_blocks: 0,
-                last_cipher_block_size: 0,
-            });
-        }
-
         let num_full_blocks = stream_length / CIPHER_BLOCK_SIZE as u64;
         let cipher_bytes_in_last_block = (stream_length % CIPHER_BLOCK_SIZE as u64) as u32;
         let full_blocks_only = cipher_bytes_in_last_block == 0;
@@ -288,7 +284,9 @@ impl FileRead for AesGcmFileRead {
     /// file. GCM authentication is verified per-block, so any tampering is detected
     /// at the granularity of individual blocks.
     async fn read(&self, range: Range<u64>) -> Result<Bytes> {
-        if range.start == range.end {
+        // An empty stream still has one block whose GCM tag must be verified, so an empty read
+        // cannot short-circuit there: that is what detects a file truncated to just its header.
+        if range.start == range.end && self.plain_stream_size != 0 {
             return Ok(Bytes::new());
         }
 
@@ -312,12 +310,8 @@ impl FileRead for AesGcmFileRead {
             ));
         }
 
-        if self.num_blocks == 0 {
-            return Ok(Bytes::new());
-        }
-
         let first_block = range.start / PLAIN_BLOCK_SIZE as u64;
-        let last_block = (range.end - 1) / PLAIN_BLOCK_SIZE as u64;
+        let last_block = range.end.saturating_sub(1) / PLAIN_BLOCK_SIZE as u64;
 
         // Read all needed encrypted blocks in a single I/O call
         let encrypted_start = Self::encrypted_block_offset(first_block);
@@ -325,6 +319,16 @@ impl FileRead for AesGcmFileRead {
             Self::encrypted_block_offset(last_block) + self.cipher_block_size(last_block) as u64;
 
         let all_encrypted = self.inner.read(encrypted_start..encrypted_end).await?;
+        if all_encrypted.len() as u64 != encrypted_end - encrypted_start {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "Invalid encrypted read length: expected {} bytes, got {}",
+                    encrypted_end - encrypted_start,
+                    all_encrypted.len()
+                ),
+            ));
+        }
 
         // Decrypt each block and extract the requested plaintext range
         let result_len = (range.end - range.start) as usize;
@@ -502,7 +506,7 @@ impl FileWrite for AesGcmFileWrite {
         Ok(())
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<FileMetadata> {
         if self.closed {
             return Err(Error::new(
                 ErrorKind::Unexpected,
@@ -640,6 +644,69 @@ mod tests {
         assert!(result.is_empty());
     }
 
+    #[tokio::test]
+    async fn test_short_ciphertext_read_is_rejected() {
+        struct ShortRead;
+
+        #[async_trait::async_trait]
+        impl FileRead for ShortRead {
+            async fn read(&self, range: Range<u64>) -> Result<Bytes> {
+                let len = (range.end - range.start).saturating_sub(1) as usize;
+                Ok(Bytes::from(vec![0; len]))
+            }
+        }
+
+        let reader = AesGcmFileRead::new(
+            Box::new(ShortRead),
+            Arc::new(make_cipher(b"0123456789abcdef")),
+            Box::default(),
+            u64::from(MIN_STREAM_LENGTH) + 10,
+        )
+        .unwrap();
+        let err = reader.read(0..10).await.unwrap_err();
+        assert_eq!(err.kind(), ErrorKind::DataInvalid);
+        assert!(err.to_string().contains("Invalid encrypted read length"));
+    }
+
+    #[tokio::test]
+    async fn test_oversized_declared_length_is_rejected() {
+        // Object stores clamp reads at EOF rather than failing, unlike the in-memory FileIO.
+        struct ClampingRead(Bytes);
+
+        #[async_trait::async_trait]
+        impl FileRead for ClampingRead {
+            async fn read(&self, range: Range<u64>) -> Result<Bytes> {
+                let start = (range.start as usize).min(self.0.len());
+                let end = (range.end as usize).min(self.0.len());
+                Ok(self.0.slice(start..end))
+            }
+        }
+
+        let key = b"0123456789abcdef";
+        let aad_prefix = b"test-aad-prefix!";
+        let plaintext = b"some bytes to measure";
+        let encrypted = write_through_ags1(plaintext, key, aad_prefix).await;
+
+        // A declared length larger than the real file must be rejected rather than yielding
+        // truncated plaintext, whether it overstates by one byte or by a whole extra block.
+        for excess in [1, u64::from(CIPHER_BLOCK_SIZE)] {
+            let reader = AesGcmFileRead::new(
+                Box::new(ClampingRead(Bytes::from(encrypted.clone()))),
+                Arc::new(make_cipher(key)),
+                aad_prefix.to_vec().into_boxed_slice(),
+                encrypted.len() as u64 + excess,
+            )
+            .unwrap();
+
+            let err = reader
+                .read(0..plaintext.len() as u64)
+                .await
+                .expect_err("an inflated declared length must not read back as plaintext");
+            assert_eq!(err.kind(), ErrorKind::DataInvalid);
+            assert!(err.to_string().contains("Invalid encrypted read length"));
+        }
+    }
+
     #[tokio::test]
     async fn test_small_file_roundtrip() {
         let key = b"0123456789abcdef";
@@ -956,15 +1023,21 @@ mod tests {
         assert_eq!(&aad[..], &42u32.to_le_bytes());
     }
 
-    #[tokio::test]
-    async fn test_encrypted_file_too_short() {
-        let result = AesGcmFileRead::new(
-            memory_reader(vec![0; 4]),
-            Arc::new(make_cipher(b"0123456789abcdef")),
-            [].into(),
-            4,
-        );
-        assert!(result.is_err());
+    #[test]
+    fn test_encrypted_file_too_short() {
+        for length in 0..MIN_STREAM_LENGTH {
+            let result = AesGcmFileRead::new(
+                memory_reader(vec![0; length as usize]),
+                Arc::new(make_cipher(b"0123456789abcdef")),
+                [].into(),
+                u64::from(length),
+            );
+            let err = result
+                .err()
+                .expect("a stream must contain an authenticated block");
+            assert_eq!(err.kind(), ErrorKind::DataInvalid);
+            assert!(err.to_string().contains("Invalid encrypted file length"));
+        }
     }
 
     // --- AesGcmFileWrite tests ---
@@ -990,8 +1063,14 @@ mod tests {
             Ok(())
         }
 
-        async fn close(&mut self) -> Result<()> {
-            Ok(())
+        // A poisoned AesGcmFileWrite must reject close() before delegating, so reaching this is a
+        // bug. Returning an error rather than panicking surfaces it as a failed assertion on the
+        // expected "poisoned" message.
+        async fn close(&mut self) -> Result<FileMetadata> {
+            Err(Error::new(
+                ErrorKind::Unexpected,
+                "FailingFileWrite::close called unexpectedly",
+            ))
         }
     }
 
@@ -1002,8 +1081,10 @@ mod tests {
             Ok(())
         }
 
-        async fn close(&mut self) -> Result<()> {
-            Ok(())
+        async fn close(&mut self) -> Result<FileMetadata> {
+            Ok(FileMetadata {
+                size: self.buffer.lock().unwrap().len() as u64,
+            })
         }
     }
 
@@ -1017,9 +1098,15 @@ mod tests {
         let mut writer = AesGcmFileWrite::new(inner, cipher, aad_prefix.to_vec());
 
         writer.write(Bytes::from(plaintext.to_vec())).await.unwrap();
-        writer.close().await.unwrap();
+        let metadata = writer.close().await.unwrap();
 
-        buffer.lock().unwrap().clone()
+        let encrypted = buffer.lock().unwrap().clone();
+        assert_eq!(
+            metadata.size,
+            encrypted.len() as u64,
+            "close() must report the full ciphertext length"
+        );
+        encrypted
     }
 
     #[tokio::test]
@@ -1239,10 +1326,9 @@ mod tests {
         );
 
         // Close should also be rejected
-        let result = writer.close().await;
-        assert!(result.is_err());
+        let err = writer.close().await.err().expect("close should fail");
         assert!(
-            result.unwrap_err().to_string().contains("poisoned"),
+            err.to_string().contains("poisoned"),
             "expected poisoned error on close"
         );
     }
diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs
index 87ac84784f..42eec4db56 100644
--- a/crates/iceberg/src/io/file_io.rs
+++ b/crates/iceberg/src/io/file_io.rs
@@ -375,10 +375,10 @@ pub trait FileWrite: Send + Unpin + 'static {
     /// TODO: we can support writing non-contiguous bytes in the future.
     async fn write(&mut self, bs: Bytes) -> Result<()>;
 
-    /// Close file.
+    /// Close the file and return its stored size, including encryption overhead for encrypted files.
     ///
     /// Calling close on closed file will generate an error.
-    async fn close(&mut self) -> Result<()>;
+    async fn close(&mut self) -> Result<FileMetadata>;
 }
 
 /// Output file is used for writing to files..
diff --git a/crates/iceberg/src/io/storage/local_fs.rs b/crates/iceberg/src/io/storage/local_fs.rs
index 836ea74d4e..f744e9b94c 100644
--- a/crates/iceberg/src/io/storage/local_fs.rs
+++ b/crates/iceberg/src/io/storage/local_fs.rs
@@ -268,12 +268,16 @@ impl FileRead for LocalFsFileRead {
 #[derive(Debug)]
 pub struct LocalFsFileWrite {
     file: Option<fs::File>,
+    bytes_written: u64,
 }
 
 impl LocalFsFileWrite {
     /// Create a new `LocalFsFileWrite` for the given file.
     pub fn new(file: fs::File) -> Self {
-        Self { file: Some(file) }
+        Self {
+            file: Some(file),
+            bytes_written: 0,
+        }
     }
 }
 
@@ -291,11 +295,12 @@ impl FileWrite for LocalFsFileWrite {
                 format!("Failed to write to file: {e}"),
             )
         })?;
+        self.bytes_written += bs.len() as u64;
 
         Ok(())
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<FileMetadata> {
         let file = self
             .file
             .take()
@@ -304,7 +309,9 @@ impl FileWrite for LocalFsFileWrite {
         file.sync_all()
             .map_err(|e| Error::new(ErrorKind::Unexpected, format!("Failed to sync file: {e}")))?;
 
-        Ok(())
+        Ok(FileMetadata {
+            size: self.bytes_written,
+        })
     }
 }
 
@@ -483,10 +490,11 @@ mod tests {
         let mut writer = storage.writer(path_str).await.unwrap();
         writer.write(Bytes::from("Hello, ")).await.unwrap();
         writer.write(Bytes::from("World!")).await.unwrap();
-        writer.close().await.unwrap();
+        let metadata = writer.close().await.unwrap();
 
         let content = storage.read(path_str).await.unwrap();
         assert_eq!(content, Bytes::from("Hello, World!"));
+        assert_eq!(metadata.size, content.len() as u64);
     }
 
     #[tokio::test]
@@ -513,7 +521,7 @@ mod tests {
         let path_str = path.to_str().unwrap();
 
         let mut writer = storage.writer(path_str).await.unwrap();
-        writer.close().await.unwrap();
+        assert_eq!(writer.close().await.unwrap().size, 0);
 
         // Write after close should fail
         let result = writer.write(Bytes::from("test")).await;
diff --git a/crates/iceberg/src/io/storage/memory.rs b/crates/iceberg/src/io/storage/memory.rs
index dc9ca10d6a..929d864a79 100644
--- a/crates/iceberg/src/io/storage/memory.rs
+++ b/crates/iceberg/src/io/storage/memory.rs
@@ -325,7 +325,7 @@ impl FileWrite for MemoryFileWrite {
         Ok(())
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn close(&mut self) -> Result<FileMetadata> {
         if self.closed {
             return Err(Error::new(ErrorKind::DataInvalid, "File already closed"));
         }
@@ -337,12 +337,13 @@ impl FileWrite for MemoryFileWrite {
             )
         })?;
 
+        let size = self.buffer.len() as u64;
         data.insert(
             self.path.clone(),
             Bytes::from(std::mem::take(&mut self.buffer)),
         );
         self.closed = true;
-        Ok(())
+        Ok(FileMetadata { size })
     }
 }
 
@@ -496,10 +497,11 @@ mod tests {
         let mut writer = storage.writer(path).await.unwrap();
         writer.write(Bytes::from("Hello, ")).await.unwrap();
         writer.write(Bytes::from("World!")).await.unwrap();
-        writer.close().await.unwrap();
+        let metadata = writer.close().await.unwrap();
 
         let content = storage.read(path).await.unwrap();
         assert_eq!(content, Bytes::from("Hello, World!"));
+        assert_eq!(metadata.size, content.len() as u64);
     }
 
     #[tokio::test]
@@ -522,7 +524,7 @@ mod tests {
         let path = "memory://test/file.txt";
 
         let mut writer = storage.writer(path).await.unwrap();
-        writer.close().await.unwrap();
+        assert_eq!(writer.close().await.unwrap().size, 0);
 
         // Write after close should fail
         let result = writer.write(Bytes::from("test")).await;
diff --git a/crates/iceberg/src/puffin/reader.rs b/crates/iceberg/src/puffin/reader.rs
index 01601f48e0..a7b178ac84 100644
--- a/crates/iceberg/src/puffin/reader.rs
+++ b/crates/iceberg/src/puffin/reader.rs
@@ -45,7 +45,7 @@ impl PuffinReader {
     /// reader operates over plaintext offsets and length, so all blob and
     /// footer positions match those written to the unencrypted file.
     pub async fn new_from_encrypted(encrypted_input: EncryptedInputFile) -> Result<Self> {
-        let file_length = encrypted_input.metadata().await?.size;
+        let file_length = encrypted_input.metadata()?.size;
         let file_read = encrypted_input.reader().await?;
         Ok(Self::from_parts(file_read, file_length))
     }
diff --git a/crates/iceberg/src/puffin/writer.rs b/crates/iceberg/src/puffin/writer.rs
index 0437bd5bf0..e6882a4844 100644
--- a/crates/iceberg/src/puffin/writer.rs
+++ b/crates/iceberg/src/puffin/writer.rs
@@ -23,7 +23,7 @@ use super::validate_puffin_compression;
 use crate::Result;
 use crate::compression::CompressionCodec;
 use crate::encryption::EncryptedOutputFile;
-use crate::io::{FileWrite, OutputFile};
+use crate::io::{FileMetadata as IoFileMetadata, FileWrite, OutputFile};
 use crate::puffin::blob::Blob;
 use crate::puffin::metadata::{BlobMetadata, FileMetadata, Flag};
 
@@ -117,12 +117,11 @@ impl PuffinWriter {
         Ok(())
     }
 
-    /// Finalizes the Puffin file
-    pub async fn close(mut self) -> Result<()> {
+    /// Finalizes the Puffin file and returns its stored size.
+    pub async fn close(mut self) -> Result<IoFileMetadata> {
         self.write_header_once().await?;
         self.write_footer().await?;
-        self.writer.close().await?;
-        Ok(())
+        self.writer.close().await
     }
 
     async fn write(&mut self, bytes: Bytes) -> Result<()> {
@@ -409,7 +408,7 @@ mod tests {
         for blob in blobs.clone() {
             writer.add(blob, CompressionCodec::None).await.unwrap();
         }
-        writer.close().await.unwrap();
+        let file_metadata = writer.close().await.unwrap();
 
         // The ciphertext on disk must not equal a plaintext puffin file.
         let raw = file_io.new_input(path).unwrap().read().await.unwrap();
@@ -419,8 +418,10 @@ mod tests {
         );
 
         // Read back through the decrypting reader over plaintext offsets.
+        let key_metadata = encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata);
+        assert_eq!(key_metadata.file_length(), Some(raw.len() as u64));
         let encrypted_input =
-            EncryptedInputFile::new(file_io.new_input(path).unwrap(), key_metadata());
+            EncryptedInputFile::new(file_io.new_input(path).unwrap(), key_metadata);
         let reader = PuffinReader::new_from_encrypted(encrypted_input)
             .await
             .unwrap();
diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs
index 7256692950..0922d3ec1c 100644
--- a/crates/iceberg/src/spec/manifest/writer.rs
+++ b/crates/iceberg/src/spec/manifest/writer.rs
@@ -16,8 +16,6 @@
 // under the License.
 
 use std::cmp::min;
-use std::future::Future;
-use std::pin::Pin;
 
 use apache_avro::{Writer as AvroWriter, to_value};
 use bytes::Bytes;
@@ -30,7 +28,7 @@ use super::{
 };
 use crate::encryption::EncryptedOutputFile;
 use crate::error::Result;
-use crate::io::{FileWrite, OutputFile};
+use crate::io::{FileMetadata, FileWrite, OutputFile};
 use crate::spec::manifest::_serde::{ManifestEntryV1, ManifestEntryV2};
 use crate::spec::manifest::{manifest_schema_v1, manifest_schema_v2};
 use crate::spec::{
@@ -43,14 +41,38 @@ use crate::{Error, ErrorKind};
 /// with the actual snapshot ID before it is committed.
 const UNASSIGNED_SNAPSHOT_ID: i64 = -1;
 
-type WriterFuture = Pin<Box<dyn Future<Output = Result<Box<dyn FileWrite>>> + Send>>;
+/// Retains the output until close provides the size required by encrypted key metadata.
+pub(crate) enum ManifestOutput {
+    Plain(OutputFile),
+    Encrypted(EncryptedOutputFile),
+}
+
+impl ManifestOutput {
+    async fn writer(&self) -> Result<Box<dyn FileWrite>> {
+        match self {
+            Self::Plain(output) => output.writer().await,
+            Self::Encrypted(output) => output.writer().await,
+        }
+    }
+
+    fn encoded_key_metadata(&self, file_metadata: &FileMetadata) -> Result<Option<Vec<u8>>> {
+        match self {
+            Self::Plain(_) => Ok(None),
+            Self::Encrypted(output) => Ok(Some(
+                output
+                    .key_metadata_with_saved_file_metadata(file_metadata)
+                    .encode()?
+                    .into_vec(),
+            )),
+        }
+    }
+}
 
 /// The builder used to create a [`ManifestWriter`].
 pub struct ManifestWriterBuilder {
-    writer_future: WriterFuture,
+    output: ManifestOutput,
     location: String,
     snapshot_id: Option<i64>,
-    key_metadata: Option<Vec<u8>>,
     schema: SchemaRef,
     partition_spec: PartitionSpec,
 }
@@ -65,10 +87,9 @@ impl ManifestWriterBuilder {
     ) -> Self {
         let location = output.location().to_owned();
         Self {
-            writer_future: Box::pin(async move { output.writer().await }),
+            output: ManifestOutput::Plain(output),
             location,
             snapshot_id,
-            key_metadata: None,
             schema,
             partition_spec,
         }
@@ -84,12 +105,10 @@ impl ManifestWriterBuilder {
         partition_spec: PartitionSpec,
     ) -> Result<Self> {
         let location = encrypted_output.location().to_owned();
-        let key_metadata = Some(encrypted_output.key_metadata().encode()?.to_vec());
         Ok(Self {
-            writer_future: Box::pin(async move { encrypted_output.writer().await }),
+            output: ManifestOutput::Encrypted(encrypted_output),
             location,
             snapshot_id,
-            key_metadata,
             schema,
             partition_spec,
         })
@@ -104,14 +123,7 @@ impl ManifestWriterBuilder {
             .format_version(FormatVersion::V1)
             .content(ManifestContentType::Data)
             .build();
-        ManifestWriter::new(
-            self.writer_future,
-            self.location,
-            self.snapshot_id,
-            self.key_metadata,
-            metadata,
-            None,
-        )
+        ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None)
     }
 
     /// Build a [`ManifestWriter`] for format version 2, data content.
@@ -124,14 +136,7 @@ impl ManifestWriterBuilder {
             .content(ManifestContentType::Data)
             .build();
 
-        ManifestWriter::new(
-            self.writer_future,
-            self.location,
-            self.snapshot_id,
-            self.key_metadata,
-            metadata,
-            None,
-        )
+        ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None)
     }
 
     /// Build a [`ManifestWriter`] for format version 2, deletes content.
@@ -143,14 +148,7 @@ impl ManifestWriterBuilder {
             .format_version(FormatVersion::V2)
             .content(ManifestContentType::Deletes)
             .build();
-        ManifestWriter::new(
-            self.writer_future,
-            self.location,
-            self.snapshot_id,
-            self.key_metadata,
-            metadata,
-            None,
-        )
+        ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None)
     }
 
     /// Build a [`ManifestWriter`] for format version 2, data content.
@@ -163,10 +161,9 @@ impl ManifestWriterBuilder {
             .content(ManifestContentType::Data)
             .build();
         ManifestWriter::new(
-            self.writer_future,
+            self.output,
             self.location,
             self.snapshot_id,
-            self.key_metadata,
             metadata,
             // First row id is assigned by the [`ManifestListWriter`] when the manifest
             // is added to the list.
@@ -183,20 +180,13 @@ impl ManifestWriterBuilder {
             .format_version(FormatVersion::V3)
             .content(ManifestContentType::Deletes)
             .build();
-        ManifestWriter::new(
-            self.writer_future,
-            self.location,
-            self.snapshot_id,
-            self.key_metadata,
-            metadata,
-            None,
-        )
+        ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None)
     }
 }
 
 /// A manifest writer.
 pub struct ManifestWriter {
-    writer_future: WriterFuture,
+    output: ManifestOutput,
     location: String,
 
     snapshot_id: Option<i64>,
@@ -211,8 +201,6 @@ pub struct ManifestWriter {
 
     min_seq_num: Option<i64>,
 
-    key_metadata: Option<Vec<u8>>,
-
     manifest_entries: Vec<ManifestEntry>,
 
     metadata: ManifestMetadata,
@@ -221,15 +209,14 @@ pub struct ManifestWriter {
 impl ManifestWriter {
     /// Create a new manifest writer.
     pub(crate) fn new(
-        writer_future: WriterFuture,
+        output: ManifestOutput,
         location: String,
         snapshot_id: Option<i64>,
-        key_metadata: Option<Vec<u8>>,
         metadata: ManifestMetadata,
         first_row_id: Option<u64>,
     ) -> Self {
         Self {
-            writer_future,
+            output,
             location,
             snapshot_id,
             added_files: 0,
@@ -240,7 +227,6 @@ impl ManifestWriter {
             deleted_rows: 0,
             first_row_id,
             min_seq_num: None,
-            key_metadata,
             manifest_entries: Vec::new(),
             metadata,
         }
@@ -508,14 +494,15 @@ impl ManifestWriter {
         }
 
         let content = avro_writer.into_inner()?;
-        let length = content.len();
-        let mut writer = self.writer_future.await?;
+        let mut writer = self.output.writer().await?;
         writer.write(Bytes::from(content)).await?;
-        writer.close().await?;
+        let file_metadata = writer.close().await?;
+        let key_metadata = self.output.encoded_key_metadata(&file_metadata)?;
 
         Ok(ManifestFile {
             manifest_path: self.location,
-            manifest_length: length as i64,
+            // Manifest lengths are on-disk sizes, including encryption overhead.
+            manifest_length: file_metadata.size.try_into()?,
             partition_spec_id: self.metadata.partition_spec.spec_id(),
             content: self.metadata.content,
             // sequence_number and min_sequence_number with UNASSIGNED_SEQUENCE_NUMBER will be replace with
@@ -530,7 +517,7 @@ impl ManifestWriter {
             existing_rows_count: Some(self.existing_rows),
             deleted_rows_count: Some(self.deleted_rows),
             partitions: Some(partition_summary),
-            key_metadata: self.key_metadata,
+            key_metadata,
             first_row_id: self.first_row_id,
         })
     }
@@ -836,9 +823,9 @@ mod tests {
         assert_eq!(manifest_file.content, ManifestContentType::Deletes);
 
         // Read back the manifest file
-        let actual_manifest =
-            Manifest::parse_avro(fs::read(&path).expect("read_file must succeed").as_slice())
-                .unwrap();
+        let bytes = fs::read(&path).expect("read_file must succeed");
+        assert_eq!(manifest_file.manifest_length, bytes.len() as i64);
+        let actual_manifest = Manifest::parse_avro(&bytes).unwrap();
 
         // Verify the content type is correctly preserved as Deletes
         assert_eq!(
diff --git a/crates/iceberg/src/spec/manifest_list/manifest_file.rs b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
index 567e78e3a5..3c7d5d9c10 100644
--- a/crates/iceberg/src/spec/manifest_list/manifest_file.rs
+++ b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
@@ -290,12 +290,15 @@ mod test {
         let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
             .unwrap()
             .with_aad_prefix(b"test-aad-prefix!");
-        let encoded_key_metadata = key_metadata.encode().unwrap().to_vec();
-
         let io = FileIO::new_with_memory();
         let path = "memory:///test/encrypted_manifest.avro";
-        let manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
-        assert_eq!(manifest_file.key_metadata, Some(encoded_key_metadata));
+        let manifest_file = write_encrypted_manifest(&io, path, key_metadata.clone()).await;
+        let size = io.new_input(path).unwrap().metadata().await.unwrap().size;
+        assert_eq!(manifest_file.manifest_length, size as i64);
+        assert_eq!(
+            StandardKeyMetadata::decode(manifest_file.key_metadata.as_ref().unwrap()).unwrap(),
+            key_metadata.with_file_length(size)
+        );
 
         let manifest = ManifestReader::new(io).read(&manifest_file).await.unwrap();
         assert_eq!(manifest.entries().len(), 1);
@@ -322,7 +325,8 @@ mod test {
         // returning garbage.
         let wrong_key_metadata = StandardKeyMetadata::try_new(b"fedcba9876543210")
             .unwrap()
-            .with_aad_prefix(b"test-aad-prefix!");
+            .with_aad_prefix(b"test-aad-prefix!")
+            .with_file_length(manifest_file.manifest_length as u64);
         manifest_file.key_metadata = Some(wrong_key_metadata.encode().unwrap().to_vec());
 
         let err = ManifestReader::new(io)
@@ -347,7 +351,8 @@ mod test {
         // so GCM authentication must fail even though the key is right.
         let wrong_aad_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
             .unwrap()
-            .with_aad_prefix(b"wrong-aad-prefix");
+            .with_aad_prefix(b"wrong-aad-prefix")
+            .with_file_length(manifest_file.manifest_length as u64);
         manifest_file.key_metadata = Some(wrong_aad_metadata.encode().unwrap().to_vec());
 
         let err = ManifestReader::new(io)
diff --git a/crates/iceberg/src/spec/manifest_list/writer.rs b/crates/iceberg/src/spec/manifest_list/writer.rs
index 710471780b..6850fe8d37 100644
--- a/crates/iceberg/src/spec/manifest_list/writer.rs
+++ b/crates/iceberg/src/spec/manifest_list/writer.rs
@@ -26,7 +26,7 @@ use super::_const_schema::{
 use super::_serde::{ManifestFileV1, ManifestFileV2, ManifestFileV3};
 use super::{FormatVersion, ManifestContentType, ManifestFile, UNASSIGNED_SEQUENCE_NUMBER};
 use crate::error::Result;
-use crate::io::FileWrite;
+use crate::io::{FileMetadata, FileWrite};
 use crate::{Error, ErrorKind};
 
 /// A manifest list writer.
@@ -196,12 +196,11 @@ impl ManifestListWriter {
         Ok(())
     }
 
-    /// Write the manifest list to the output file.
-    pub async fn close(mut self) -> Result<()> {
+    /// Write the manifest list and return its stored size.
+    pub async fn close(mut self) -> Result<FileMetadata> {
         let data = self.avro_writer.into_inner()?;
         self.writer.write(Bytes::from(data)).await?;
-        self.writer.close().await?;
-        Ok(())
+        self.writer.close().await
     }
 
     /// Assign sequence numbers to manifest if they are unassigned
@@ -616,7 +615,6 @@ mod test {
         let path = "memory:///manifest_list_v3_encrypted.avro";
 
         let encrypted_output = mgr.encrypt(file_io.new_output(path).unwrap());
-        let key_metadata = encrypted_output.key_metadata().clone();
 
         let snapshot_id = 9_000_000_000_000_001i64;
         let seq_num = 7i64;
@@ -651,7 +649,7 @@ mod test {
         writer
             .add_manifests(expected.entries.clone().into_iter())
             .unwrap();
-        writer.close().await.unwrap();
+        let file_metadata = writer.close().await.unwrap();
 
         let raw_bytes = file_io.new_input(path).unwrap().read().await.unwrap();
         assert!(
@@ -659,6 +657,8 @@ mod test {
             "raw bytes should be ciphertext, not parseable as Avro"
         );
 
+        let key_metadata = encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata);
+        assert_eq!(key_metadata.file_length(), Some(raw_bytes.len() as u64));
         let plaintext = EncryptedInputFile::new(file_io.new_input(path).unwrap(), key_metadata)
             .read()
             .await
diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs
index 872a1d89c1..3260f9ffa2 100644
--- a/crates/iceberg/src/transaction/append.rs
+++ b/crates/iceberg/src/transaction/append.rs
@@ -420,6 +420,25 @@ mod tests {
             })
             .expect("a fast append should emit an AddSnapshot update");
 
+        let manifest_list_key_metadata = table
+            .encryption_manager()
+            .unwrap()
+            .decrypt_manifest_list_key_metadata(new_snapshot.encryption_key_id().unwrap())
+            .await
+            .unwrap();
+        let manifest_list_size = table
+            .file_io()
+            .new_input(new_snapshot.manifest_list())
+            .unwrap()
+            .metadata()
+            .await
+            .unwrap()
+            .size;
+        assert_eq!(
+            manifest_list_key_metadata.file_length(),
+            Some(manifest_list_size)
+        );
+
         let manifest_list = table
             .manifest_list_reader(&new_snapshot)
             .load()
@@ -436,8 +455,18 @@ mod tests {
             .key_metadata
             .as_ref()
             .expect("encrypted manifest must record key metadata");
-        StandardKeyMetadata::decode(key_metadata_bytes)
+        let key_metadata = StandardKeyMetadata::decode(key_metadata_bytes)
             .expect("recorded key metadata must decode as StandardKeyMetadata");
+        let manifest_size = table
+            .file_io()
+            .new_input(&manifest_file.manifest_path)
+            .unwrap()
+            .metadata()
+            .await
+            .unwrap()
+            .size;
+        assert_eq!(key_metadata.file_length(), Some(manifest_size));
+        assert_eq!(manifest_file.manifest_length, manifest_size as i64);
 
         // The reader self-decrypts using the recorded key metadata and must
         // recover the entry we appended. Because the read goes through the
diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs
index 1f09a79a08..e4b4f370fd 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -451,13 +451,13 @@ impl<'a> SnapshotProducer<'a> {
             .file_io()
             .new_output(manifest_list_path.clone())?;
 
-        let (writer, encryption_key_id) = match self.table.encryption_manager() {
+        let (writer, encrypted_output) = match self.table.encryption_manager() {
             Some(em) => {
                 let encrypted_output = em.encrypt(raw_output);
-                let key_id = em
-                    .encrypt_manifest_list_key_metadata(encrypted_output.key_metadata())
-                    .await?;
-                (encrypted_output.writer().await?, Some(key_id))
+                (
+                    encrypted_output.writer().await?,
+                    Some((em.clone(), encrypted_output)),
+                )
             }
             None => (raw_output.writer().await?, None),
         };
@@ -492,7 +492,16 @@ impl<'a> SnapshotProducer<'a> {
 
         manifest_list_writer.add_manifests(new_manifests.into_iter())?;
         let writer_next_row_id = manifest_list_writer.next_row_id();
-        manifest_list_writer.close().await?;
+        let file_metadata = manifest_list_writer.close().await?;
+        let encryption_key_id = match encrypted_output {
+            Some((em, encrypted_output)) => Some(
+                em.encrypt_manifest_list_key_metadata(
+                    &encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata),
+                )
+                .await?,
+            ),
+            None => None,
+        };
 
         let commit_ts = chrono::Utc::now().timestamp_millis();
         let new_snapshot = Snapshot::builder()
diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs
index fbf333c7bf..ebbd4e9c7e 100644
--- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs
+++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs
@@ -743,9 +743,11 @@ impl ArrowAsyncFileWriter for AsyncFileWriter {
 
     fn complete(&mut self) -> BoxFuture<'_, parquet::errors::Result<()>> {
         Box::pin(async {
+            // TODO(encryption): retain the stored file size in data-file key metadata.
             self.0
                 .close()
                 .await
+                .map(|_| ())
                 .map_err(|err| parquet::errors::ParquetError::External(Box::new(err)))
         })
     }
diff --git a/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json b/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json
index 0dc0aabc17..1c096417fb 100644
--- a/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json
+++ b/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json
@@ -63,7 +63,7 @@
     },
     {
       "key-id": "1fb95f04-7f98-4b1f-a08f-e810cfb907ff",
-      "encrypted-key-metadata": "scAmTJzgxf7vDQBetLMWSKnZmOljEzoitb6H90HQ/wjDz5/naIkg44PkAg+Ldl6RBP0FBcoS1bfBDVLToUbdrVA=",
+      "encrypted-key-metadata": "XcrqLxxyf9a7FAFXM63u3MEH8CSWqh6g1H8fJD2/6zx0EjcPzPaP/5AtRpO2s6p2isV8s7a4m77zj+GIqNmIpbA2Xw==",
       "encrypted-by-id": "5b29941b-b3b4-4d57-8758-a9d1bf340b39"
     }
   ],
diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs
index 2ed5a60cb1..fe485cb915 100644
--- a/crates/storage/opendal/src/lib.rs
+++ b/crates/storage/opendal/src/lib.rs
@@ -551,7 +551,7 @@ impl Storage for OpenDalStorage {
 
     async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
         let (op, relative_path) = self.create_operator(&path)?;
-        Ok(Box::new(OpenDalWriter(
+        Ok(Box::new(OpenDalWriter::new(
             op.writer(relative_path).await.map_err(from_opendal_error)?,
         )))
     }
@@ -635,21 +635,51 @@ impl FileRead for OpenDalReader {
 }
 
 /// Wrapper around `opendal::Writer` that implements `FileWrite`.
-pub(crate) struct OpenDalWriter(pub(crate) opendal::Writer);
+pub(crate) struct OpenDalWriter {
+    inner: opendal::Writer,
+    bytes_written: u64,
+}
+
+impl OpenDalWriter {
+    pub(crate) fn new(inner: opendal::Writer) -> Self {
+        Self {
+            inner,
+            bytes_written: 0,
+        }
+    }
+}
 
 #[async_trait]
 impl FileWrite for OpenDalWriter {
     async fn write(&mut self, bs: Bytes) -> Result<()> {
-        Ok(opendal::Writer::write(&mut self.0, bs)
+        let len = bs.len() as u64;
+        opendal::Writer::write(&mut self.inner, bs)
             .await
-            .map_err(from_opendal_error)?)
+            .map_err(from_opendal_error)?;
+        self.bytes_written += len;
+        Ok(())
     }
 
-    async fn close(&mut self) -> Result<()> {
-        let _ = opendal::Writer::close(&mut self.0)
+    async fn close(&mut self) -> Result<FileMetadata> {
+        let metadata = opendal::Writer::close(&mut self.inner)
             .await
             .map_err(from_opendal_error)?;
-        Ok(())
+
+        // Object stores may omit the size (reported as 0); validate only a reported nonzero size.
+        let reported_size = metadata.content_length();
+        if reported_size != 0 && reported_size != self.bytes_written {
+            return Err(Error::new(
+                ErrorKind::Unexpected,
+                format!(
+                    "Wrote {} bytes but storage reports {reported_size}",
+                    self.bytes_written
+                ),
+            ));
+        }
+
+        Ok(FileMetadata {
+            size: self.bytes_written,
+        })
     }
 }
 
@@ -695,6 +725,38 @@ mod tests {
         assert_eq!(op.info().scheme().to_string(), "memory");
     }
 
+    #[cfg(feature = "opendal-memory")]
+    #[tokio::test]
+    async fn test_writer_close_returns_stored_size() {
+        use iceberg::encryption::{EncryptedOutputFile, StandardKeyMetadata};
+
+        // Note: the memory service does report a content length, so this only pins the happy
+        // path. The counter in `OpenDalWriter` is what covers services that don't, such as S3.
+        let storage = Arc::new(OpenDalStorage::Memory(default_memory_operator()));
+        let path = "memory:///stored-size";
+        for plaintext in [
+            Bytes::new(),
+            Bytes::from_static(b"test data"),
+            Bytes::from(vec![7; 3 * 1024]),
+        ] {
+            let mut writer = storage.writer(path).await.unwrap();
+            for chunk in plaintext.chunks(1024) {
+                writer.write(Bytes::copy_from_slice(chunk)).await.unwrap();
+            }
+            let metadata = writer.close().await.unwrap();
+            assert_eq!(metadata.size, plaintext.len() as u64);
+            assert_eq!(metadata.size, storage.metadata(path).await.unwrap().size);
+
+            let output = EncryptedOutputFile::new(
+                OutputFile::new(storage.clone(), path.to_string()),
+                StandardKeyMetadata::try_new(b"0123456789abcdef").unwrap(),
+            );
+            let metadata = output.write(plaintext.clone()).await.unwrap();
+            assert!(metadata.size > plaintext.len() as u64);
+            assert_eq!(metadata.size, storage.metadata(path).await.unwrap().size);
+        }
+    }
+
     #[cfg(feature = "opendal-memory")]
     #[test]
     fn test_relativize_path_memory() {