iceberg maintenance: cache a remote-only file before reading it (#12191)

* iceberg maintenance: an unreadable parquet file fails the job, not the process

parquet.NewReader panics on any file it cannot open (wrong magic, a
truncated footer, zero bytes), and parquet-go's row decoder panics on a
corrupt page under a valid footer. Maintenance runs in the worker's
goroutines with no recover, so one bad delete or data file in any table
ended the whole process.

- Every parquet open in compaction goes through newParquetReader, which
  opens with parquet.OpenFile and returns the error.
- A panic while decoding a delete file or a data file's rows is recovered
  into an error naming the file (recoverRead/guardRead).
- As a backstop, detection recovers per table and Execute per operation:
  the table is skipped or the operation fails, and the worker keeps
  running.

A failed delete file still fails the table's compaction (no commit), as a
read error did before; a failed data file still skips only its bin.

* iceberg maintenance: cache a remote-only file before reading it

On a bucket mounted with remote.mount, a file that is not cached locally
has no chunks and no inline content, only its RemoteEntry. Maintenance
read it as an empty file: a parquet reader failed on zero bytes (before
the previous change, it panicked and ended the process) and a manifest
failed to parse, so a table whose files had been uncached could not be
maintained.

lookupFileByIcebergPath, the lookup every read goes through, now calls
CacheRemoteObjectToLocalCluster for a remote-only entry, as the S3
gateway does on a read, and reads the entry it returns; the call is tried
twice, and an entry still remote-only after caching is an error. The
lookups that only resolve a path (orphan removal) do not cache.
readEntryContent reports a file with no content and no chunks as empty
instead of returning zero bytes.

The fake volume server in the tests now serves chunk reads, so the
remote-only tests read the cached file through the chunk reader as a
real one is read.

* iceberg maintenance: bound the metadata location read before caching a remote file

* iceberg maintenance: bound the metadata read on the entry the cache returns

* iceberg maintenance: enforce the metadata size bound inside the remote cache call

The worker checks the resolved entry before asking the filer to cache a
remote-only object, but the filer resolves the entry again at cache time
and downloads whatever its remote size then says. A replacement landing in
between could push an oversized object through the download even though
the post-cache check would refuse to read it.

CacheRemoteObjectToLocalClusterRequest gains an optional max_size; the
filer rejects an over-limit remote object before the download starts.
loadMetadata passes metadataFileMaxBytes; data-file reads leave it unset.

* iceberg maintenance: key the remote cache singleflight by the size limit

Callers sharing one flight share its outcome, including a bounded
caller's over-limit rejection: an unrestricted reader joining the same
path would fail even though its own bound allows the download. The
limit joins the key so only callers with the same bound share a flight.

* iceberg: gofmt compact_order_test.go

* filer: bound remote cache callers before the shared download

Each caller's max_size is now enforced on the entry the filer resolves
before that caller joins the singleflight, which is keyed by path again.
A bounded caller's rejection is never shared with other callers, and
callers with different bounds share one download instead of pulling the
same remote object twice.

* filer: bind the remote cache size bound to the downloaded entry

The pre-flight check and the shared download resolve the entry
separately, so a remote metadata refresh landing in between lets a
larger object through: the download ran to the refreshed size and only
the reader's post-check refused it. The bound is now re-checked inside
the flight on the entry the download streams.

The rejection is typed so a co-tenant sharing the flight does not
inherit it: callers whose own bound permits the reported size retry on
a fresh flight instead of failing for another caller's limit.

---------

Co-authored-by: user.email <chrislusf@users.noreply.github.com>

weed/pb/filer.proto

@@ -823,6 +823,7 @@ message CacheRemoteObjectToLocalClusterRequest {
     string name = 2;
     int32 chunk_concurrency = 3; // parallel chunk downloads per file, 0 = default (8)
     int32 download_concurrency = 4; // multipart download concurrency per chunk (if supported by remote storage), 0 = default (5 for S3)
+    int64 max_size = 5; // reject remote objects larger than this before downloading, 0 = no limit
 }
 message CacheRemoteObjectToLocalClusterResponse {
     Entry entry = 1;

weed/pb/filer_pb/filer.pb.go

@@ -5447,6 +5447,7 @@ type CacheRemoteObjectToLocalClusterRequest struct {
 	Name                string                 `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
 	ChunkConcurrency    int32                  `protobuf:"varint,3,opt,name=chunk_concurrency,json=chunkConcurrency,proto3" json:"chunk_concurrency,omitempty"`          // parallel chunk downloads per file, 0 = default (8)
 	DownloadConcurrency int32                  `protobuf:"varint,4,opt,name=download_concurrency,json=downloadConcurrency,proto3" json:"download_concurrency,omitempty"` // multipart download concurrency per chunk (if supported by remote storage), 0 = default (5 for S3)
+	MaxSize             int64                  `protobuf:"varint,5,opt,name=max_size,json=maxSize,proto3" json:"max_size,omitempty"`                                     // reject remote objects larger than this before downloading, 0 = no limit
 	unknownFields       protoimpl.UnknownFields
 	sizeCache           protoimpl.SizeCache
 }
@@ -5509,6 +5510,13 @@ func (x *CacheRemoteObjectToLocalClusterRequest) GetDownloadConcurrency() int32
 	return 0
 }
 
+func (x *CacheRemoteObjectToLocalClusterRequest) GetMaxSize() int64 {
+	if x != nil {
+		return x.MaxSize
+	}
+	return 0
+}
+
 type CacheRemoteObjectToLocalClusterResponse struct {
 	state         protoimpl.MessageState     `protogen:"open.v1"`
 	Entry         *Entry                     `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"`
@@ -7872,12 +7880,13 @@ const file_filer_proto_rawDesc = "" +
 	"\x19worm_grace_period_seconds\x18\x0f \x01(\x04R\x16wormGracePeriodSeconds\x12=\n" +
 	"\x1bworm_retention_time_seconds\x18\x10 \x01(\x04R\x18wormRetentionTimeSecondsB\b\n" +
 	"\x06_fsyncB\a\n" +
-	"\x05_worm\"\xba\x01\n" +
+	"\x05_worm\"\xd5\x01\n" +
 	"&CacheRemoteObjectToLocalClusterRequest\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" +
 	"\x04name\x18\x02 \x01(\tR\x04name\x12+\n" +
 	"\x11chunk_concurrency\x18\x03 \x01(\x05R\x10chunkConcurrency\x121\n" +
-	"\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\"\xdd\x01\n" +
+	"\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\x12\x19\n" +
+	"\bmax_size\x18\x05 \x01(\x03R\amaxSize\"\xdd\x01\n" +
 	"'CacheRemoteObjectToLocalClusterResponse\x12%\n" +
 	"\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12J\n" +
 	"\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\x12\x1a\n" +

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -5173,6 +5173,11 @@ func (m *CacheRemoteObjectToLocalClusterRequest) MarshalToSizedBufferVT(dAtA []b
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.MaxSize != 0 {
+		i = protohelpers.EncodeVarint(dAtA, i, uint64(m.MaxSize))
+		i--
+		dAtA[i] = 0x28
+	}
 	if m.DownloadConcurrency != 0 {
 		i = protohelpers.EncodeVarint(dAtA, i, uint64(m.DownloadConcurrency))
 		i--
@@ -8683,6 +8688,9 @@ func (m *CacheRemoteObjectToLocalClusterRequest) SizeVT() (n int) {
 	if m.DownloadConcurrency != 0 {
 		n += 1 + protohelpers.SizeOfVarint(uint64(m.DownloadConcurrency))
 	}
+	if m.MaxSize != 0 {
+		n += 1 + protohelpers.SizeOfVarint(uint64(m.MaxSize))
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -23123,6 +23131,25 @@ func (m *CacheRemoteObjectToLocalClusterRequest) UnmarshalVT(dAtA []byte) error
 					break
 				}
 			}
+		case 5:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field MaxSize", wireType)
+			}
+			m.MaxSize = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.MaxSize |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])