[#XX] container: Add ListStream method

Signed-off-by: Ekaterina Lebedeva <ekaterina.lebedeva@yadro.com>

.golangci.yml

@@ -50,7 +50,7 @@ linters:
     - bidichk
     - durationcheck
     - exhaustive
-    - exportloopref
+    - copyloopvar
     - gofmt
     - goimports
     - misspell

container/convert.go

@@ -762,3 +762,138 @@ func (r *ListResponse) FromGRPCMessage(m grpc.Message) error {
 
 	return r.ResponseHeaders.FromMessage(v)
 }
+
+func (r *ListStreamRequestBody) ToGRPCMessage() grpc.Message {
+	var m *container.ListStreamRequest_Body
+
+	if r != nil {
+		m = new(container.ListStreamRequest_Body)
+
+		m.SetOwnerId(r.ownerID.ToGRPCMessage().(*refsGRPC.OwnerID))
+	}
+
+	return m
+}
+
+func (r *ListStreamRequestBody) FromGRPCMessage(m grpc.Message) error {
+	v, ok := m.(*container.ListStreamRequest_Body)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, v)
+	}
+
+	var err error
+
+	ownerID := v.GetOwnerId()
+	if ownerID == nil {
+		r.ownerID = nil
+	} else {
+		if r.ownerID == nil {
+			r.ownerID = new(refs.OwnerID)
+		}
+
+		err = r.ownerID.FromGRPCMessage(ownerID)
+	}
+
+	return err
+}
+
+func (r *ListStreamRequest) ToGRPCMessage() grpc.Message {
+	var m *container.ListStreamRequest
+
+	if r != nil {
+		m = new(container.ListStreamRequest)
+
+		m.SetBody(r.body.ToGRPCMessage().(*container.ListStreamRequest_Body))
+		r.RequestHeaders.ToMessage(m)
+	}
+
+	return m
+}
+
+func (r *ListStreamRequest) FromGRPCMessage(m grpc.Message) error {
+	v, ok := m.(*container.ListStreamRequest)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, v)
+	}
+
+	var err error
+
+	body := v.GetBody()
+	if body == nil {
+		r.body = nil
+	} else {
+		if r.body == nil {
+			r.body = new(ListStreamRequestBody)
+		}
+
+		err = r.body.FromGRPCMessage(body)
+		if err != nil {
+			return err
+		}
+	}
+
+	return r.RequestHeaders.FromMessage(v)
+}
+
+func (r *ListStreamResponseBody) ToGRPCMessage() grpc.Message {
+	var m *container.ListStreamResponse_Body
+
+	if r != nil {
+		m = new(container.ListStreamResponse_Body)
+
+		m.SetContainerIds(refs.ContainerIDsToGRPCMessage(r.cidList))
+	}
+
+	return m
+}
+
+func (r *ListStreamResponseBody) FromGRPCMessage(m grpc.Message) error {
+	v, ok := m.(*container.ListStreamResponse_Body)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, v)
+	}
+
+	var err error
+
+	r.cidList, err = refs.ContainerIDsFromGRPCMessage(v.GetContainerIds())
+
+	return err
+}
+
+func (r *ListStreamResponse) ToGRPCMessage() grpc.Message {
+	var m *container.ListStreamResponse
+
+	if r != nil {
+		m = new(container.ListStreamResponse)
+
+		m.SetBody(r.body.ToGRPCMessage().(*container.ListStreamResponse_Body))
+		r.ResponseHeaders.ToMessage(m)
+	}
+
+	return m
+}
+
+func (r *ListStreamResponse) FromGRPCMessage(m grpc.Message) error {
+	v, ok := m.(*container.ListStreamResponse)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, v)
+	}
+
+	var err error
+
+	body := v.GetBody()
+	if body == nil {
+		r.body = nil
+	} else {
+		if r.body == nil {
+			r.body = new(ListStreamResponseBody)
+		}
+
+		err = r.body.FromGRPCMessage(body)
+		if err != nil {
+			return err
+		}
+	}
+
+	return r.ResponseHeaders.FromMessage(v)
+}

container/grpc/service_frostfs_fuzz.go

@@ -157,3 +157,41 @@ func DoFuzzJSONListResponse(data []byte) int {
 	}
 	return 1
 }
+func DoFuzzProtoListStreamRequest(data []byte) int {
+	msg := new(ListStreamRequest)
+	if err := msg.UnmarshalProtobuf(data); err != nil {
+		return 0
+	}
+	_ = msg.MarshalProtobuf(nil)
+	return 1
+}
+func DoFuzzJSONListStreamRequest(data []byte) int {
+	msg := new(ListStreamRequest)
+	if err := msg.UnmarshalJSON(data); err != nil {
+		return 0
+	}
+	_, err := msg.MarshalJSON()
+	if err != nil {
+		panic(err)
+	}
+	return 1
+}
+func DoFuzzProtoListStreamResponse(data []byte) int {
+	msg := new(ListStreamResponse)
+	if err := msg.UnmarshalProtobuf(data); err != nil {
+		return 0
+	}
+	_ = msg.MarshalProtobuf(nil)
+	return 1
+}
+func DoFuzzJSONListStreamResponse(data []byte) int {
+	msg := new(ListStreamResponse)
+	if err := msg.UnmarshalJSON(data); err != nil {
+		return 0
+	}
+	_, err := msg.MarshalJSON()
+	if err != nil {
+		panic(err)
+	}
+	return 1
+}

container/grpc/service_frostfs_test.go

@@ -89,3 +89,23 @@ func FuzzJSONListResponse(f *testing.F) {
 		DoFuzzJSONListResponse(data)
 	})
 }
+func FuzzProtoListStreamRequest(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		DoFuzzProtoListStreamRequest(data)
+	})
+}
+func FuzzJSONListStreamRequest(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		DoFuzzJSONListStreamRequest(data)
+	})
+}
+func FuzzProtoListStreamResponse(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		DoFuzzProtoListStreamResponse(data)
+	})
+}
+func FuzzJSONListStreamResponse(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		DoFuzzJSONListStreamResponse(data)
+	})
+}

container/marshal.go

@@ -343,3 +343,65 @@ func (r *ListResponseBody) StableSize() (size int) {
 func (r *ListResponseBody) Unmarshal(data []byte) error {
 	return message.Unmarshal(r, data, new(container.ListResponse_Body))
 }
+
+func (r *ListStreamRequestBody) StableMarshal(buf []byte) []byte {
+	if r == nil {
+		return []byte{}
+	}
+
+	if buf == nil {
+		buf = make([]byte, r.StableSize())
+	}
+
+	protoutil.NestedStructureMarshal(listReqBodyOwnerField, buf, r.ownerID)
+
+	return buf
+}
+
+func (r *ListStreamRequestBody) StableSize() (size int) {
+	if r == nil {
+		return 0
+	}
+
+	size += protoutil.NestedStructureSize(listReqBodyOwnerField, r.ownerID)
+
+	return size
+}
+
+func (r *ListStreamRequestBody) Unmarshal(data []byte) error {
+	return message.Unmarshal(r, data, new(container.ListStreamRequest_Body))
+}
+
+func (r *ListStreamResponseBody) StableMarshal(buf []byte) []byte {
+	if r == nil {
+		return []byte{}
+	}
+
+	if buf == nil {
+		buf = make([]byte, r.StableSize())
+	}
+
+	var offset int
+
+	for i := range r.cidList {
+		offset += protoutil.NestedStructureMarshal(listRespBodyIDsField, buf[offset:], &r.cidList[i])
+	}
+
+	return buf
+}
+
+func (r *ListStreamResponseBody) StableSize() (size int) {
+	if r == nil {
+		return 0
+	}
+
+	for i := range r.cidList {
+		size += protoutil.NestedStructureSize(listRespBodyIDsField, &r.cidList[i])
+	}
+
+	return size
+}
+
+func (r *ListStreamResponseBody) Unmarshal(data []byte) error {
+	return message.Unmarshal(r, data, new(container.ListStreamResponse_Body))
+}

container/types.go

@@ -109,6 +109,26 @@ type ListResponse struct {
 	session.ResponseHeaders
 }
 
+type ListStreamRequestBody struct {
+	ownerID *refs.OwnerID
+}
+
+type ListStreamRequest struct {
+	body *ListStreamRequestBody
+
+	session.RequestHeaders
+}
+
+type ListStreamResponseBody struct {
+	cidList []refs.ContainerID
+}
+
+type ListStreamResponse struct {
+	body *ListStreamResponseBody
+
+	session.ResponseHeaders
+}
+
 func (a *Attribute) GetKey() string {
 	if a != nil {
 		return a.key
@@ -444,3 +464,51 @@ func (r *ListResponse) GetBody() *ListResponseBody {
 func (r *ListResponse) SetBody(v *ListResponseBody) {
 	r.body = v
 }
+
+func (r *ListStreamRequestBody) GetOwnerID() *refs.OwnerID {
+	if r != nil {
+		return r.ownerID
+	}
+
+	return nil
+}
+
+func (r *ListStreamRequestBody) SetOwnerID(v *refs.OwnerID) {
+	r.ownerID = v
+}
+
+func (r *ListStreamRequest) GetBody() *ListStreamRequestBody {
+	if r != nil {
+		return r.body
+	}
+
+	return nil
+}
+
+func (r *ListStreamRequest) SetBody(v *ListStreamRequestBody) {
+	r.body = v
+}
+
+func (r *ListStreamResponseBody) GetContainerIDs() []refs.ContainerID {
+	if r != nil {
+		return r.cidList
+	}
+
+	return nil
+}
+
+func (r *ListStreamResponseBody) SetContainerIDs(v []refs.ContainerID) {
+	r.cidList = v
+}
+
+func (r *ListStreamResponse) GetBody() *ListStreamResponseBody {
+	if r != nil {
+		return r.body
+	}
+
+	return nil
+}
+
+func (r *ListStreamResponse) SetBody(v *ListStreamResponseBody) {
+	r.body = v
+}

rpc/container.go

@@ -13,6 +13,7 @@ const (
 	rpcContainerGet       = "Get"
 	rpcContainerDel       = "Delete"
 	rpcContainerList      = "List"
+	rpcContainerStream    = "ListStream"
 	rpcContainerGetEACL   = "GetExtendedACL"
 	rpcContainerUsedSpace = "AnnounceUsedSpace"
 )
@@ -80,3 +81,27 @@ func ListContainers(
 
 	return resp, nil
 }
+
+type ListStreamResponseReader struct {
+	r client.MessageReader
+}
+
+func (r *ListStreamResponseReader) Read(resp *container.ListStreamResponse) error {
+	return r.r.ReadMessage(resp)
+}
+
+// ListContainersStream executes ContainerService.ListStream RPC.
+func ListContainersStream(
+	cli *client.Client,
+	req *container.ListStreamRequest,
+	opts ...client.CallOption,
+) (*ListStreamResponseReader, error) {
+	wc, err := client.OpenServerStream(cli, common.CallMethodInfoServerStream(serviceContainer, rpcContainerList), req, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ListStreamResponseReader{
+		r: wc,
+	}, nil
+}

signature/body.go

@@ -46,6 +46,10 @@ func serviceMessageBody(req any) stableMarshaler {
 		return v.GetBody()
 	case *container.ListResponse:
 		return v.GetBody()
+	case *container.ListStreamRequest:
+		return v.GetBody()
+	case *container.ListStreamResponse:
+		return v.GetBody()
 
 		/* Object */
 	case *object.PutRequest:

util/proto/marshal_test.go

@@ -26,12 +26,35 @@ func nonZero[T protoInt]() T {
 
 func TestStableMarshalSingle(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
+		t.Run("proto", func(t *testing.T) {
 			input := &generated.Primitives{}
 			require.Zero(t, input.StableSize())
 
 			r := input.MarshalProtobuf(nil)
 			require.Empty(t, r)
 		})
+		t.Run("json", func(t *testing.T) {
+			input := &generated.Primitives{}
+			r, err := input.MarshalJSON()
+			require.NoError(t, err)
+			require.NotEmpty(t, r)
+
+			var actual test.Primitives
+			require.NoError(t, protojson.Unmarshal(r, &actual))
+
+			t.Run("protojson compatibility", func(t *testing.T) {
+				data, err := protojson.MarshalOptions{EmitUnpopulated: true}.Marshal(&actual)
+				require.NoError(t, err)
+				require.JSONEq(t, string(data), string(r))
+			})
+
+			var actualFrostfs generated.Primitives
+			require.NoError(t, actualFrostfs.UnmarshalJSON(r))
+			require.Equal(t, input, &actualFrostfs)
+
+			primitivesEqual(t, input, &actual)
+		})
+	})
 
 	marshalCases := []struct {
 		name  string
@@ -77,8 +100,7 @@ func TestStableMarshalSingle(t *testing.T) {
 				require.NoError(t, protojson.Unmarshal(r, &actual))
 
 				t.Run("protojson compatibility", func(t *testing.T) {
-					t.Skip()
-					data, err := protojson.Marshal(&actual)
+					data, err := protojson.MarshalOptions{EmitUnpopulated: true}.Marshal(&actual)
 					require.NoError(t, err)
 					require.JSONEq(t, string(data), string(r))
 				})
@@ -95,6 +117,7 @@ func TestStableMarshalSingle(t *testing.T) {
 
 func primitivesEqual(t *testing.T, a *generated.Primitives, b *test.Primitives) {
 	// Compare each field directly, because proto-generated code has private fields.
+	require.Equal(t, len(a.FieldA), len(b.FieldA))
 	require.Equal(t, a.FieldA, b.FieldA)
 	require.Equal(t, a.FieldB, b.FieldB)
 	require.Equal(t, a.FieldC, b.FieldC)

util/protogen/internalgengo/json.go

@@ -59,6 +59,35 @@ func emitJSONUnmarshal(g *protogen.GeneratedFile, msg *protogen.Message) {
 	g.P("}")
 }
 
+func emitJSONParseInteger(g *protogen.GeneratedFile, ident string, method string, bitSize int, typ string) {
+	g.P("r := in.JsonNumber()")
+	g.P("n := r.String()")
+	g.P("v, err := ", strconvPackage.Ident(method), "(n, 10, ", bitSize, ")")
+	g.P("if err != nil {")
+	g.P("	in.AddError(err)")
+	g.P("	return")
+	g.P("}")
+	g.P(ident, " := ", typ, "(v)")
+}
+
+func emitJSONReadEnum(g *protogen.GeneratedFile, name string, enumType string) {
+	g.P(`switch v := in.Interface().(type) {
+	case string:
+		if vv, ok := `+enumType+`_value[v]; ok {
+			`+name+` = `+enumType+`(vv)
+			break
+		}
+		vv, err := `, strconvPackage.Ident("ParseInt"), `(v, 10, 32)
+		if err != nil {
+			in.AddError(err)
+			return
+		}
+		`+name+` = `+enumType+`(vv)
+	case float64:
+		`+name+` = `+enumType+`(v)
+	}`)
+}
+
 func emitJSONFieldRead(g *protogen.GeneratedFile, f *protogen.Field, name string) {
 	g.P("{")
 	defer g.P("}")
@@ -83,30 +112,20 @@ func emitJSONFieldRead(g *protogen.GeneratedFile, f *protogen.Field, name string
 
 		enumType := fieldType(g, f).String()
 		g.P("var parsedValue " + enumType)
-		g.P(`switch v := in.Interface().(type) {
-		case string:
-			if vv, ok := `+enumType+`_value[v]; ok {
-				parsedValue = `+enumType+`(vv)
-				break
-			}
-			vv, err := `, strconvPackage.Ident("ParseInt"), `(v, 10, 32)
-			if err != nil {
-				in.AddError(err)
-				return
-			}
-			parsedValue = `+enumType+`(vv)
-		case float64:
-			parsedValue = `+enumType+`(v)
-		}`)
+		emitJSONReadEnum(g, "parsedValue", enumType)
 		template = "%s = parsedValue"
 	case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind:
-		template = "%s = in.Int32()"
+		emitJSONParseInteger(g, "pv", "ParseInt", 32, "int32")
+		template = "%s = pv"
 	case protoreflect.Uint32Kind, protoreflect.Fixed32Kind:
-		template = "%s = in.Uint32()"
+		emitJSONParseInteger(g, "pv", "ParseUint", 32, "uint32")
+		template = "%s = pv"
 	case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
-		template = "%s = in.Int64()"
+		emitJSONParseInteger(g, "pv", "ParseInt", 64, "int64")
+		template = "%s = pv"
 	case protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
-		template = "%s = in.Uint64()"
+		emitJSONParseInteger(g, "pv", "ParseUint", 64, "uint64")
+		template = "%s = pv"
 	case protoreflect.FloatKind:
 		template = "%s = in.Float32()"
 	case protoreflect.DoubleKind:
@@ -114,7 +133,17 @@ func emitJSONFieldRead(g *protogen.GeneratedFile, f *protogen.Field, name string
 	case protoreflect.StringKind:
 		template = "%s = in.String()"
 	case protoreflect.BytesKind:
-		template = "%s = in.Bytes()"
+		// Since some time ago proto3 support optional keyword, thus the presence is not tracked by default:
+		// https://github.com/protocolbuffers/protobuf-go/blob/fb995f184a1719ec42b247a3771d1036d92adf67/internal/impl/message_reflect_field.go#L327
+		// We do not explicitly support `optional` keyword, protoc will fail on such fileds.
+		// Thus, treat empty string as `[]byte(nil)`.
+		template = `{
+			tmp := in.Bytes()
+			if len(tmp) == 0 {
+				tmp = nil
+			}
+			%s = tmp
+		}`
 	case protoreflect.MessageKind:
 		if f.Desc.IsList() {
 			g.P("f = ", fieldType(g, f)[2:], "{}")
@@ -147,8 +176,11 @@ func emitJSONMarshal(g *protogen.GeneratedFile, msg *protogen.Message) {
 	g.P("func (x *", msg.GoIdent.GoName, ") MarshalEasyJSON(out *", jwriterPackage.Ident("Writer"), ") {")
 	g.P(`if x == nil { out.RawString("null"); return }`)
 
+	if len(msg.Fields) != 0 {
+		g.P("first := true")
+	}
 	g.P("out.RawByte('{')")
-	for i, f := range msg.Fields {
+	for _, f := range msg.Fields {
 		if f.Oneof != nil {
 			if f.Oneof.Fields[0] != f {
 				continue
@@ -157,29 +189,38 @@ func emitJSONMarshal(g *protogen.GeneratedFile, msg *protogen.Message) {
 			g.P("switch xx := x.", f.Oneof.GoName, ".(type) {")
 			for _, ff := range f.Oneof.Fields {
 				g.P("case *", ff.GoIdent, ":")
-				emitJSONFieldWrite(g, ff, "xx", i == 0)
+				emitJSONFieldWrite(g, ff, "xx")
 			}
 			g.P("}")
 			continue
 		}
-		emitJSONFieldWrite(g, f, "x", i == 0)
+		emitJSONFieldWrite(g, f, "x")
 	}
 	g.P("out.RawByte('}')")
 	g.P("}")
 }
 
-func emitJSONFieldWrite(g *protogen.GeneratedFile, f *protogen.Field, name string, first bool) {
+func emitJSONFieldWrite(g *protogen.GeneratedFile, f *protogen.Field, name string) {
 	g.P("{")
 	defer g.P("}")
 
-	g.P("const prefix string = ", `",\"`, fieldJSONName(f), `\":"`)
-	if first {
-		g.P("out.RawString(prefix[1:])")
-	} else {
-		g.P("out.RawString(prefix)")
-	}
-
 	selector := name + "." + f.GoName
+
+	// This code is responsible for ignoring default values.
+	// We will restore it after having parametrized JSON marshaling.
+	//
+	// isNotDefault := notNil
+	// if f.Desc.IsList() {
+	// 	isNotDefault = notEmpty
+	// } else if f.Desc.Kind() != protoreflect.MessageKind {
+	// 	_, isNotDefault = easyprotoKindInfo(f.Desc.Kind())
+	// }
+	// g.P("if ", isNotDefault(selector), "{")
+	// defer g.P("}")
+
+	g.P("if !first { out.RawByte(','); } else { first = false; }")
+	g.P("const prefix string = ", `"\"`, fieldJSONName(f), `\":"`)
+	g.P("out.RawString(prefix)")
 	if f.Desc.IsList() {
 		selector += "[i]"
 		g.P("out.RawByte('[')")
@@ -195,15 +236,27 @@ func emitJSONFieldWrite(g *protogen.GeneratedFile, f *protogen.Field, name strin
 	case protoreflect.BoolKind:
 		template = "out.Bool(%s)"
 	case protoreflect.EnumKind:
-		template = "out.Int32(int32(%s))"
+		enumType := fieldType(g, f).String()
+		template = `v := int32(%s)
+			if vv, ok := ` + enumType + `_name[v]; ok {
+				out.String(vv)
+			} else {
+				out.Int32(v)
+			}`
 	case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind:
 		template = "out.Int32(%s)"
 	case protoreflect.Uint32Kind, protoreflect.Fixed32Kind:
 		template = "out.Uint32(%s)"
 	case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
-		template = "out.Int64(%s)"
+		g.P("out.RawByte('\"')")
+		g.P("out.Buffer.Buf = ", strconvPackage.Ident("AppendInt"), "(out.Buffer.Buf, ", selector, ", 10)")
+		g.P("out.RawByte('\"')")
+		return
 	case protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
-		template = "out.Uint64(%s)"
+		g.P("out.RawByte('\"')")
+		g.P("out.Buffer.Buf = ", strconvPackage.Ident("AppendUint"), "(out.Buffer.Buf, ", selector, ", 10)")
+		g.P("out.RawByte('\"')")
+		return
 	case protoreflect.FloatKind:
 		template = "out.Float32(%s)"
 	case protoreflect.DoubleKind:
@@ -211,7 +264,10 @@ func emitJSONFieldWrite(g *protogen.GeneratedFile, f *protogen.Field, name strin
 	case protoreflect.StringKind:
 		template = "out.String(%s)"
 	case protoreflect.BytesKind:
-		template = "out.Base64Bytes(%s)"
+		g.P("if ", selector, "!= nil {")
+		g.P("out.Base64Bytes(", selector, ")")
+		g.P("} else { out.String(\"\") }")
+		return
 	case protoreflect.MessageKind:
 		template = "%s.MarshalEasyJSON(out)"
 	case protoreflect.GroupKind: