fix(fileservice): validate remote cache payloads (#25969)

The remote-cache RPC boundary trusted request shape and response indexes.
Schema-valid requests with an empty key list or an omitted nested `CacheKey`
could panic the serving CN. A malformed or inconsistent response could use a
negative/out-of-range index, an index requested from a different target, or an
incorrect data length to panic the requester or mark invalid bytes as a hit.

Validate every request entry before allocating the I/O vector, require all
keys in one request to name the same file, and return `bad request` without
installing a response release callback on invalid input. If `ReadCache`
partially acquires cache data and then fails, release it immediately.

On the client, accept only indexes sent to that target, reject out-of-bounds
and wrong-length hits, and let the normal local-storage path handle every
invalid response. An invalid hit does not consume its index, while the first
fully valid response wins. The MORPC response is released on all received
response paths.

Normal same-version senders already generate valid shapes; nil repeated
message pointers are only possible for direct in-process callers. The network-
relevant cases are empty/missing nested request fields and invalid scalar
response indexes/lengths.

Validation:

- focused fileservice regressions: PASS
- focused fileservice regressions with `-race -count=10`: PASS
- `.agents/skills/mo-dev/scripts/mo-cgo-test -count=1 -timeout=180s ./pkg/fileservice`
- `.agents/skills/mo-dev/scripts/mo-cgo-test -race -count=1 -timeout=240s ./pkg/fileservice`
- `.agents/skills/mo-dev/scripts/mo-cgo-test -count=1 -timeout=240s ./pkg/cnservice`
- `go build ./pkg/fileservice ./pkg/cnservice` (repository CGo environment)
- `go vet ./pkg/fileservice ./pkg/cnservice` (repository CGo environment)

Approved by: @gouhongshen, @XuPeng-SH

Author: LeftHandCold

pkg/cnservice/server_query_test.go

@@ -1187,7 +1187,6 @@ func Test_service_handleGetCacheData(t *testing.T) {
 	require.NoError(t, err)
 
 	mockQuery := mock_query.NewMockQueryService(ctl)
-	mockQuery.EXPECT().SetReleaseFunc(gomock.Any(), gomock.Any()).Return().AnyTimes()
 
 	type fields struct {
 		fileService  fileservice.FileService
@@ -1230,7 +1229,7 @@ func Test_service_handleGetCacheData(t *testing.T) {
 			want:    &query.Response{},
 		},
 		{
-			name: "read_empty",
+			name: "missing_cache_key",
 			fields: fields{
 				fileService:  fs,
 				queryService: mockQuery,
@@ -1249,7 +1248,7 @@ func Test_service_handleGetCacheData(t *testing.T) {
 				},
 				resp: &query.Response{},
 			},
-			wantErr: nil,
+			wantErr: dummyBadRequestErr,
 			want:    &query.Response{GetCacheDataResponse: nil},
 		},
 	}

pkg/fileservice/remote_cache.go

@@ -114,26 +114,47 @@ func (r *RemoteCache) Read(ctx context.Context, vector *IOVector) error {
 		req.GetCacheDataRequest = &query.GetCacheDataRequest{
 			RequestCacheKey: key,
 		}
+		requestedIndexes := make(map[int]struct{}, len(key))
+		for _, cacheKey := range key {
+			requestedIndexes[int(cacheKey.Index)] = struct{}{}
+		}
 
 		func(ctx context.Context) {
 			ctx, cancel := context.WithTimeoutCause(ctx, time.Second*2, moerr.CauseRemoteCacheRead)
 			defer cancel()
 			resp, err := r.client.SendMessage(ctx, target, req)
-			if err != nil {
+			if err != nil || resp == nil {
 				// Do not return error here to read data from local storage.
 				return
 			}
 			defer r.client.Release(resp)
 			if resp.GetCacheDataResponse != nil {
+				seen := make(map[int]struct{}, len(requestedIndexes))
 				for _, cacheData := range resp.GetCacheDataResponse.ResponseCacheData {
-					numRead++
+					if cacheData == nil || cacheData.Index < 0 {
+						continue
+					}
 					idx := int(cacheData.Index)
+					if idx >= len(vector.Entries) {
+						continue
+					}
+					if _, ok := requestedIndexes[idx]; !ok {
+						continue
+					}
+					if _, ok := seen[idx]; ok {
+						continue
+					}
 					if cacheData.Hit {
+						if int64(len(cacheData.Data)) != vector.Entries[idx].Size {
+							continue
+						}
 						vector.Entries[idx].done = true
 						vector.Entries[idx].CachedData = &Bytes{bytes: cacheData.Data}
 						vector.Entries[idx].fromCache = r
 						numHit++
 					}
+					seen[idx] = struct{}{}
+					numRead++
 				}
 			}
 		}(ctx)
@@ -162,27 +183,32 @@ func (r *RemoteCache) Close(ctx context.Context) {
 func HandleRemoteRead(
 	ctx context.Context, fs FileService, req *query.Request, resp *query.WrappedResponse,
 ) error {
-	if req.GetCacheDataRequest == nil {
+	if req == nil || req.GetCacheDataRequest == nil {
 		return moerr.NewInternalError(ctx, "bad request")
 	}
-	first := req.GetCacheDataRequest.RequestCacheKey[0].CacheKey
-	if first == nil { // We cannot get the first one.
-		return nil
+	keys := req.GetCacheDataRequest.RequestCacheKey
+	if len(keys) == 0 || keys[0] == nil || keys[0].CacheKey == nil {
+		return moerr.NewInternalError(ctx, "bad request")
 	}
+	first := keys[0].CacheKey
 
 	ioVec := &IOVector{
 		FilePath: first.Path,
 	}
-	ioVec.Entries = make([]IOEntry, len(req.GetCacheDataRequest.RequestCacheKey))
-	for i, k := range req.GetCacheDataRequest.RequestCacheKey {
+	ioVec.Entries = make([]IOEntry, len(keys))
+	for i, k := range keys {
+		if k == nil || k.CacheKey == nil || k.CacheKey.Path != first.Path {
+			return moerr.NewInternalError(ctx, "bad request")
+		}
 		ioVec.Entries[i].Offset = k.CacheKey.Offset
 		ioVec.Entries[i].Size = k.CacheKey.Sz
 	}
 	if err := fs.ReadCache(ctx, ioVec); err != nil {
+		ioVec.Release()
 		return err
 	}
-	respData := make([]*query.ResponseCacheData, len(req.GetCacheDataRequest.RequestCacheKey))
-	for i, k := range req.GetCacheDataRequest.RequestCacheKey {
+	respData := make([]*query.ResponseCacheData, len(keys))
+	for i, k := range keys {
 		var data []byte
 		if ioVec.Entries[i].CachedData != nil {
 			data = ioVec.Entries[i].CachedData.Bytes()