frostfs-node/pkg/services/object/search/v2/request_forwarder.go

package searchsvc

import (
	"context"
	"crypto/ecdsa"
	"errors"
	"fmt"
	"io"
	"sync"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal"
	objectV2 "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/object"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc"
	rpcclient "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc/client"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/session"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/signature"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	healthV1 "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/status"
)

type requestForwarder struct {
	OnceResign sync.Once
	Request    *objectV2.SearchRequest
	Key        *ecdsa.PrivateKey
}

func rpcSearchDialer(ctx context.Context, cc grpc.ClientConnInterface) error {
	healthClient := healthV1.NewHealthClient(cc)
	resp, err := healthClient.Check(ctx, &healthV1.HealthCheckRequest{})
	if err != nil {
		if status.Code(err) == codes.Unimplemented {
			return nil
		}
		return fmt.Errorf("health check error: %w", err)
	}
	if resp.GetStatus() != healthV1.HealthCheckResponse_SERVING {
		return errors.New("unavailable grpc server")
	}
	return nil
}

func (f *requestForwarder) forwardRequest(ctx context.Context, addr network.Address, c client.MultiAddressClient, pubkey []byte) ([]oid.ID, error) {
	var err error

	// once compose and resign forwarding request
	f.OnceResign.Do(func() {
		// compose meta header of the local server
		metaHdr := new(session.RequestMetaHeader)
		metaHdr.SetTTL(f.Request.GetMetaHeader().GetTTL() - 1)
		// TODO: #1165 think how to set the other fields
		metaHdr.SetOrigin(f.Request.GetMetaHeader())

		f.Request.SetMetaHeader(metaHdr)

		err = signature.SignServiceMessage(f.Key, f.Request)
	})

	if err != nil {
		return nil, err
	}

	var searchStream *rpc.SearchResponseReader
	err = c.RawForAddress(ctx, addr, func(cli *rpcclient.Client) error {
		searchStream, err = rpc.SearchObjects(cli, f.Request, rpcclient.WithContext(ctx), rpcclient.WithDialer(rpcSearchDialer))
		return err
	})
	if err != nil {
		return nil, err
	}

	// code below is copy-pasted from c.SearchObjects implementation,
	// perhaps it is worth highlighting the utility function in frostfs-api-go
	var (
		searchResult []oid.ID
		resp         = new(objectV2.SearchResponse)
	)

	for {
		// receive message from server stream
		err := searchStream.Read(resp)
		if err != nil {
			if errors.Is(err, io.EOF) {
				break
			}

			return nil, fmt.Errorf("reading the response failed: %w", err)
		}

		// verify response key
		if err = internal.VerifyResponseKeyV2(pubkey, resp); err != nil {
			return nil, err
		}

		// verify response structure
		if err := signature.VerifyServiceMessage(resp); err != nil {
			return nil, fmt.Errorf("could not verify %T: %w", resp, err)
		}

		chunk := resp.GetBody().GetIDList()
		var id oid.ID

		for i := range chunk {
			err = id.ReadFromV2(chunk[i])
			if err != nil {
				return nil, fmt.Errorf("invalid object ID: %w", err)
			}

			searchResult = append(searchResult, id)
		}
	}

	return searchResult, nil
}