ale64bit/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#620] object: Send status response for server-side streams

Browse source 
Previously status responses were wrapped in the gRPC error and thus
couldn't be correctly handled on client.

Introduced in c2617baf63, thanks @ale64bit for having found.

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
Evgenii Stratonikov 2023-08-17 14:24:38 +03:00 committed by Evgenii Stratonikov
parent 6407bb5bd1
commit 5a51b78946
1 changed files with 30 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
pkg/services/object/sign.go
View file

@ -49,7 +49,11 @@ func NewSignService(key *ecdsa.PrivateKey, svc ServiceServer) *SignService {
} }
func (s *getStreamSigner) Send(resp *object.GetResponse) error { func (s *getStreamSigner) Send(resp *object.GetResponse) error {
if err := s.sigSvc.SignResponse(resp, nil); err != nil { return s.send(resp, nil)
}
func (s *getStreamSigner) send(resp *object.GetResponse, err error) error {
if err := s.sigSvc.SignResponse(resp, err); err != nil {
return err return err
} }
return s.GetObjectStream.Send(resp) return s.GetObjectStream.Send(resp)
@ -62,10 +66,14 @@ func (s *SignService) Get(req *object.GetRequest, stream GetObjectStream) error
return stream.Send(resp) return stream.Send(resp)
} }
return s.svc.Get(req, &getStreamSigner{ w := &getStreamSigner{
GetObjectStream: stream, GetObjectStream: stream,
sigSvc: s.sigSvc, sigSvc: s.sigSvc,
}) }
if err := s.svc.Get(req, w); err != nil {
return w.send(new(object.GetResponse), err)
}
return nil
} }
func (s *putStreamSigner) Send(ctx context.Context, req *object.PutRequest) error { func (s *putStreamSigner) Send(ctx context.Context, req *object.PutRequest) error {
@ -124,7 +132,11 @@ func (s *SignService) PutSingle(ctx context.Context, req *object.PutSingleReques
func (s *searchStreamSigner) Send(resp *object.SearchResponse) error { func (s *searchStreamSigner) Send(resp *object.SearchResponse) error {
s.nonEmptyResp = true s.nonEmptyResp = true
if err := s.sigSvc.SignResponse(resp, nil); err != nil { return s.send(resp, nil)
}
func (s *searchStreamSigner) send(resp *object.SearchResponse, err error) error {
if err := s.sigSvc.SignResponse(resp, err); err != nil {
return err return err
} }
return s.SearchStream.Send(resp) return s.SearchStream.Send(resp)
@ -142,15 +154,15 @@ func (s *SignService) Search(req *object.SearchRequest, stream SearchStream) err
sigSvc: s.sigSvc, sigSvc: s.sigSvc,
} }
err := s.svc.Search(req, ss) err := s.svc.Search(req, ss)
if err == nil && !ss.nonEmptyResp { if err != nil || !ss.nonEmptyResp {
// The higher component does not write any response in the case of an empty result (which is correct). // The higher component does not write any response in the case of an empty result (which is correct).
// With the introduction of status returns at least one answer must be signed and sent to the client. // With the introduction of status returns at least one answer must be signed and sent to the client.
// This approach is supported by clients who do not know how to work with statuses (one could make // This approach is supported by clients who do not know how to work with statuses (one could make
// a switch according to the protocol version from the request, but the costs of sending an empty // a switch according to the protocol version from the request, but the costs of sending an empty
// answer can be neglected due to the gradual refusal to use the "old" clients). // answer can be neglected due to the gradual refusal to use the "old" clients).
return stream.Send(new(object.SearchResponse)) return ss.send(new(object.SearchResponse), err)
} }
return err return nil
} }
func (s *SignService) Delete(ctx context.Context, req *object.DeleteRequest) (*object.DeleteResponse, error) { func (s *SignService) Delete(ctx context.Context, req *object.DeleteRequest) (*object.DeleteResponse, error) {
@ -163,7 +175,11 @@ func (s *SignService) Delete(ctx context.Context, req *object.DeleteRequest) (*o
} }
func (s *getRangeStreamSigner) Send(resp *object.GetRangeResponse) error { func (s *getRangeStreamSigner) Send(resp *object.GetRangeResponse) error {
if err := s.sigSvc.SignResponse(resp, nil); err != nil { return s.send(resp, nil)
}
func (s *getRangeStreamSigner) send(resp *object.GetRangeResponse, err error) error {
if err := s.sigSvc.SignResponse(resp, err); err != nil {
return err return err
} }
return s.GetObjectRangeStream.Send(resp) return s.GetObjectRangeStream.Send(resp)
@ -176,10 +192,14 @@ func (s *SignService) GetRange(req *object.GetRangeRequest, stream GetObjectRang
return stream.Send(resp) return stream.Send(resp)
} }
return s.svc.GetRange(req, &getRangeStreamSigner{ w := &getRangeStreamSigner{
GetObjectRangeStream: stream, GetObjectRangeStream: stream,
sigSvc: s.sigSvc, sigSvc: s.sigSvc,
}) }
if err := s.svc.GetRange(req, w); err != nil {
return w.send(new(object.GetRangeResponse), err)
}
return nil
} }
func (s *SignService) GetRangeHash(ctx context.Context, req *object.GetRangeHashRequest) (*object.GetRangeHashResponse, error) { func (s *SignService) GetRangeHash(ctx context.Context, req *object.GetRangeHashRequest) (*object.GetRangeHashResponse, error) {