fix(encryption) [18/N] wire in ags1 file length for tamper proofing

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

crates/iceberg/public-api.txt

@@ -273,15 +273,15 @@ 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
 pub async fn iceberg::encryption::EncryptedInputFile::exists(&self) -> iceberg::Result<bool>
 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

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).