fix resource leak (#68)

* test build

* handle unknown OpCode

* rm test panic

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/bufbuild/protovalidate-go v0.4.3
 	github.com/go-http-utils/headers v0.0.0-20181008091004-fed159eddc2a
 	github.com/gonvenience/ytbx v1.4.4
-	github.com/google/uuid v1.4.0
+	github.com/google/uuid v1.5.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0
 	github.com/homeport/dyff v1.6.0
 	github.com/magefile/mage v1.15.0
@@ -23,7 +23,7 @@ require (
 	github.com/samber/lo v1.39.0
 	github.com/stretchr/testify v1.8.4
 	go.etcd.io/bbolt v1.3.8
-	google.golang.org/grpc v1.59.0
+	google.golang.org/grpc v1.60.0
 	google.golang.org/protobuf v1.31.0
 )
 

go.sum

@@ -71,6 +71,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/pprof v0.0.0-20230602150820-91b7bce49751 h1:hR7/MlvK23p6+lIw9SN1TigNLn9ZnF3W4SYRKq2gAHs=
 github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
 github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
+github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 h1:UH//fgunKIs4JdUbpDl1VZCDaL56wXCB/5+wF6uHfaI=
 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0/go.mod h1:g5qyo/la0ALbONm6Vbp88Yd8NsDy6rZz+RcrMPxvld8=
@@ -245,6 +247,8 @@ google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8
 google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk=
 google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
 google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
+google.golang.org/grpc v1.60.0 h1:6FQAR0kM31P6MRdeluor2w2gPaS4SVNrD/DNTxrQ15k=
+google.golang.org/grpc v1.60.0/go.mod h1:OlCHIeLYqSSsLi6i49B5QGdzaMZK9+M7LXN2FKz4eGM=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=

pkg/bdb/generic.go

@@ -1,12 +1,11 @@
 package bdb
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 
-	"github.com/aserto-dev/go-directory/pkg/pb"
 	bolt "go.etcd.io/bbolt"
+	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -67,18 +66,23 @@ func Delete(ctx context.Context, tx *bolt.Tx, path Path, key string) error {
 }
 
 func Marshal[T any, M Message[T]](t M) ([]byte, error) {
-	buf := new(bytes.Buffer)
-	if err := pb.ProtoToBuf(buf, any(t).(proto.Message)); err != nil {
-		return nil, err
-	}
-	return buf.Bytes(), nil
+	return protojson.MarshalOptions{
+		Multiline:       false,
+		Indent:          "",
+		AllowPartial:    false,
+		UseProtoNames:   true,
+		UseEnumNumbers:  false,
+		EmitUnpopulated: false,
+	}.Marshal(any(t).(proto.Message))
 }
 
 func Unmarshal[T any, M Message[T]](b []byte) (M, error) {
 	var t T
-	if err := pb.BufToProto(bytes.NewReader(b), any(&t).(proto.Message)); err != nil {
+
+	if err := (protojson.UnmarshalOptions{DiscardUnknown: true}.Unmarshal(b, any(&t).(proto.Message))); err != nil {
 		return nil, err
 	}
+
 	return &t, nil
 }
 

pkg/directory/v3/importer.go

@@ -25,6 +25,7 @@ type Importer struct {
 
 func NewImporter(logger *zerolog.Logger, store *bdb.BoltDB) *Importer {
 	v, _ := protovalidate.New()
+
 	return &Importer{
 		logger: logger,
 		store:  store,
@@ -45,21 +46,29 @@ func (s *Importer) Import(stream dsi3.Importer_ImportServer) error {
 
 	importErr := s.store.DB().Batch(func(tx *bolt.Tx) error {
 		for {
+			select {
+			case <-ctx.Done(): // exit if context is done
+				return nil
+			default:
+			}
+
 			req, err := stream.Recv()
 			if err == io.EOF {
-				s.logger.Debug().Interface("res", res).Msg("import stream EOF")
-				return stream.Send(res)
-			} else if err != nil {
-				s.logger.Err(err).Msg("cannot receive req")
+				s.logger.Trace().Interface("res", res).Msg("import stream EOF")
 				return stream.Send(res)
 			}
 
+			if err != nil {
+				s.logger.Trace().Str("err", err.Error()).Msg("cannot receive req")
+				continue
+			}
+
 			if err := s.handleImportRequest(ctx, tx, req, res); err != nil {
 				s.logger.Err(err).Msg("cannot handle load request")
-				return stream.Send(res)
 			}
 		}
 	})
+
 	return importErr
 }
 
@@ -70,26 +79,35 @@ func (s *Importer) handleImportRequest(ctx context.Context, tx *bolt.Tx, req *ds
 		if req.OpCode == dsi3.Opcode_OPCODE_SET {
 			err = s.objectSetHandler(ctx, tx, m.Object)
 			res.Object = updateCounter(res.Object, req.OpCode, err)
+			return err
 		}
 
 		if req.OpCode == dsi3.Opcode_OPCODE_DELETE {
 			err = s.objectDeleteHandler(ctx, tx, m.Object)
 			res.Object = updateCounter(res.Object, req.OpCode, err)
+			return err
 		}
 
+		return derr.ErrUnknownOpCode.Msgf("%s - %d", req.OpCode.Enum().String, int32(req.OpCode))
+
 	case *dsi3.ImportRequest_Relation:
 		if req.OpCode == dsi3.Opcode_OPCODE_SET {
 			err = s.relationSetHandler(ctx, tx, m.Relation)
 			res.Relation = updateCounter(res.Relation, req.OpCode, err)
+			return err
 		}
 
 		if req.OpCode == dsi3.Opcode_OPCODE_DELETE {
 			err = s.relationDeleteHandler(ctx, tx, m.Relation)
 			res.Relation = updateCounter(res.Relation, req.OpCode, err)
+			return err
 		}
-	}
 
-	return err
+		return derr.ErrUnknownOpCode.Msgf("%s - %d", req.OpCode.Enum().String, int32(req.OpCode))
+
+	default:
+		return derr.ErrUnknown.Msgf("import request")
+	}
 }
 
 func (s *Importer) objectSetHandler(ctx context.Context, tx *bolt.Tx, req *dsc3.Object) error {